From b2d06428257f32e832c05567b27f1481cd9ad7e2 Mon Sep 17 00:00:00 2001 From: Nick Craig-Wood Date: Sat, 18 Jul 2026 09:26:28 +0100 Subject: [PATCH] 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 | 118 +++++++++++++++++++---- cmd/serve/s3/multipart_test.go | 169 +++++++++++++++++++++++++++++++++ cmd/serve/s3/s3.go | 19 ++-- cmd/serve/s3/serve_s3.md | 27 ++++-- 4 files changed, 299 insertions(+), 34 deletions(-) diff --git a/cmd/serve/s3/multipart.go b/cmd/serve/s3/multipart.go index f5c7d56cc..f3211341a 100644 --- a/cmd/serve/s3/multipart.go +++ b/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 - putDone chan struct{} // closed when the background PutStream returns - putErr error // PutStream result (read only after putDone is closed) - nextPart int // next part number to stream (1-based) - streamBuf map[int]*pool.RW // parts received ahead of nextPart, awaiting their turn - pumping bool // a goroutine is currently writing to the pipe + putCancel context.CancelFunc // cancels the background PutStream + putDone chan struct{} // closed when the background PutStream returns + putErr error // PutStream result (read only after putDone is closed) + nextPart int // next part number to stream (1-based) + streamBuf map[int]*pool.RW // parts received ahead of nextPart, awaiting their turn + 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 { - return &multipartUpload{ - bucket: bucket, - key: key, - fp: fp, - streamFp: streamFp, - meta: meta, - partMD5s: map[int][]byte{}, - partSizes: map[int]int64{}, - nextPart: 1, - streamBuf: map[int]*pool.RW{}, +func newMultipartUpload(bucket, key, fp, streamFp string, meta map[string]string, bufferLimit int64) *multipartUpload { + up := &multipartUpload{ + bucket: bucket, + key: key, + fp: fp, + streamFp: streamFp, + meta: meta, + partMD5s: map[int][]byte{}, + partSizes: map[int]int64{}, + nextPart: 1, + 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 { diff --git a/cmd/serve/s3/multipart_test.go b/cmd/serve/s3/multipart_test.go index 4aa812250..2807e4074 100644 --- a/cmd/serve/s3/multipart_test.go +++ b/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) { diff --git a/cmd/serve/s3/s3.go b/cmd/serve/s3/s3.go index 70d18ce99..5712b10ea 100644 --- a/cmd/serve/s3/s3.go +++ b/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"` - EtagHash string `config:"etag_hash"` - AuthKey []string `config:"auth_key"` - NoCleanup bool `config:"no_cleanup"` - DisableMultipartStreaming bool `config:"disable_multipart_streaming"` - Auth httplib.AuthConfig - HTTP httplib.Config + ForcePathStyle bool `config:"force_path_style"` + EtagHash string `config:"etag_hash"` + AuthKey []string `config:"auth_key"` + NoCleanup bool `config:"no_cleanup"` + DisableMultipartStreaming bool `config:"disable_multipart_streaming"` + MultipartStreamingBufferLimit fs.SizeSuffix `config:"multipart_streaming_buffer_limit"` + Auth httplib.AuthConfig + HTTP httplib.Config } // Opt is options set by command line flags diff --git a/cmd/serve/s3/serve_s3.md b/cmd/serve/s3/serve_s3.md index d4457cbb2..3c8f9df5a 100644 --- a/cmd/serve/s3/serve_s3.md +++ b/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 - uses more memory; non-contiguous part numbers are rejected. Configure - the client to upload in part order, ideally with low concurrency, for - the lowest memory use. -- No per-part retry. Once a part has been streamed it is committed, so a - failure partway through aborts the whole upload and the client must - start it again, rather than retrying a single part. (The remote's own - upload still retries its internal chunks.) + are buffered until their turn. The memory used for this buffering is + capped, per upload, by `--multipart-streaming-buffer-limit` (default + `256M`, `0` for no limit): a part that would take the buffer over the + limit is stalled until the stream drains, so a client that uploads + faster than the remote can accept sees backpressure rather than + unbounded server memory use. Since a stalled part holds its HTTP + request open, clients whose upload concurrency times chunk size + 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.