serve s3: stream multipart uploads to the backend instead of buffering in memory

Previously serve s3 buffered every part of a multipart upload in memory
(in the gofakes3 S3 library) and concatenated them when the upload
completed, so memory use grew with the size of the upload.

serve s3 now streams the parts, in part-number order, into a single
PutStream upload to the underlying remote, which performs its own upload
with bounded memory. The whole file is never held in memory - memory use
is bounded by the parts in flight. This works for any remote that
supports PutStream (nearly all, including crypt) and for any part size,
so clients that don't produce uniform-sized parts (for example
PostgreSQL backup tools such as pgBarman and pgBackRest) work too.

Parts must arrive in ascending, contiguous part-number order; parts
uploaded out of order are buffered until their turn, and there is no
per-part retry (a failure aborts the whole upload). These trade-offs are
documented.

Passing --disable-multipart-streaming, or using a remote without
PutStream, reverts to buffering the parts in memory (the previous
behaviour); a one-off NOTICE is logged the first time this happens.

Fixes #7453

cmd/serve/s3/backend.go

@@ -23,10 +23,20 @@ var (
 )
 
 // s3Backend implements the gofacess3.Backend interface to make an S3
-// backend for gofakes3
+// backend for gofakes3. It also implements gofakes3.MultipartBackend so that
+// multipart uploads stream straight through to the underlying Fs via
+// PutStream, instead of being buffered in memory by gofakes3.
 type s3Backend struct {
 	s    *Server
 	meta *sync.Map
+
+	// multipartUploads tracks in-flight streaming multipart uploads,
+	// keyed by gofakes3.UploadID.
+	multipartUploads sync.Map
+
+	// warnInMemoryOnce logs a single NOTICE the first time a multipart
+	// upload falls back to being buffered in memory.
+	warnInMemoryOnce sync.Once
 }
 
 // newBackend creates a new SimpleBucketBackend.

cmd/serve/s3/multipart.go

@@ -0,0 +1,388 @@
+// Multipart upload support for serve s3.
+//
+// Multipart uploads received by serve s3 are streamed, in part-number order,
+// into a single PutStream upload to the underlying Fs, so the whole file is
+// never buffered in memory. This implements the gofakes3.MultipartBackend
+// interface on s3Backend.
+//
+// When streaming is disabled (--disable-multipart-streaming) or the Fs has no
+// PutStream, ErrMultipartUploadNotSupported is returned so that gofakes3 falls
+// back to buffering the parts in memory.
+
+package s3
+
+import (
+	"context"
+	"crypto/md5"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"path"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/ncw/swift/v2"
+	"github.com/rclone/gofakes3"
+	"github.com/rclone/rclone/fs"
+	"github.com/rclone/rclone/fs/object"
+	"github.com/rclone/rclone/lib/multipart"
+	"github.com/rclone/rclone/lib/pool"
+)
+
+// multipartUpload tracks one in-flight S3 multipart upload that is being
+// streamed, in part order, into a single PutStream upload to the underlying Fs.
+type multipartUpload struct {
+	bucket, key string
+	fp          string // = path.Join(bucket, key)
+	meta        map[string]string
+
+	pipeW *io.PipeWriter // parts are streamed here, in part-number order
+
+	mu        sync.Mutex
+	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
+}
+
+// newMultipartUpload allocates an upload struct.
+func newMultipartUpload(bucket, key, fp string, meta map[string]string) *multipartUpload {
+	return &multipartUpload{
+		bucket:    bucket,
+		key:       key,
+		fp:        fp,
+		meta:      meta,
+		partMD5s:  map[int][]byte{},
+		partSizes: map[int]int64{},
+		nextPart:  1,
+		streamBuf: map[int]*pool.RW{},
+	}
+}
+
+// loadUpload looks up an in-flight upload by ID.
+func (b *s3Backend) loadUpload(uploadID gofakes3.UploadID) (*multipartUpload, error) {
+	v, ok := b.multipartUploads.Load(uploadID)
+	if !ok {
+		return nil, gofakes3.ErrNoSuchUpload
+	}
+	return v.(*multipartUpload), nil
+}
+
+// CreateMultipartUpload begins a new multipart upload that streams the parts,
+// in part-number order, into a single PutStream upload to the underlying Fs.
+//
+// If streaming is disabled (--disable-multipart-streaming) or the Fs has no
+// PutStream, ErrMultipartUploadNotSupported is returned so that gofakes3 falls
+// back to buffering the whole upload in memory; a one-off NOTICE warns about
+// the memory use.
+func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objectName string, meta map[string]string) (gofakes3.UploadID, error) {
+	_vfs, err := b.s.getVFS(ctx)
+	if err != nil {
+		return "", err
+	}
+	if _, err := _vfs.Stat(bucketName); err != nil {
+		return "", gofakes3.BucketNotFound(bucketName)
+	}
+
+	f := _vfs.Fs()
+	features := f.Features()
+	if b.s.opt.DisableMultipartStreaming || features.PutStream == nil {
+		b.warnInMemoryOnce.Do(func() {
+			reason := "this backend doesn't support streaming uploads"
+			if b.s.opt.DisableMultipartStreaming {
+				reason = "--disable-multipart-streaming is set"
+			}
+			fs.Logf(nil, "serve s3: buffering multipart uploads in memory because %s - this may use a lot of memory", reason)
+		})
+		return "", gofakes3.ErrMultipartUploadNotSupported
+	}
+
+	fp := path.Join(bucketName, objectName)
+	objectDir := path.Dir(fp)
+	if objectDir != "." {
+		if err := mkdirRecursive(objectDir, _vfs); err != nil {
+			return "", err
+		}
+	}
+
+	up := newMultipartUpload(bucketName, objectName, fp, meta)
+
+	src := object.NewStaticObjectInfo(fp, time.Now(), -1, true, nil, f)
+	pr, pw := io.Pipe()
+	// Use a context that outlives this request (it's cancelled on abort) but
+	// keeps its values.
+	putCtx, cancel := context.WithCancel(context.WithoutCancel(ctx))
+	up.pipeW = pw
+	up.putCancel = cancel
+	up.putDone = make(chan struct{})
+	go func() {
+		_, err := features.PutStream(putCtx, pr, src)
+		up.putErr = err
+		_ = pr.CloseWithError(err)
+		close(up.putDone)
+	}()
+
+	uploadID := gofakes3.UploadID(uuid.New().String())
+	b.multipartUploads.Store(uploadID, up)
+	return uploadID, nil
+}
+
+// UploadPart writes a single part from the S3 client into the streaming upload.
+func (b *s3Backend) UploadPart(ctx context.Context, bucketName, objectName string, uploadID gofakes3.UploadID, partNumber int, contentLength int64, body io.Reader) (string, error) {
+	up, err := b.loadUpload(uploadID)
+	if 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)
+	hasher := md5.New()
+	n, err := io.Copy(rw, io.TeeReader(body, hasher))
+	if err != nil {
+		_ = rw.Close()
+		return "", err
+	}
+	if n != contentLength {
+		_ = rw.Close()
+		return "", gofakes3.ErrIncompleteBody
+	}
+	md5Sum := hasher.Sum(nil)
+	etag := fmt.Sprintf("%q", hex.EncodeToString(md5Sum))
+
+	if err := up.streamPart(partNumber, n, md5Sum, rw); err != nil {
+		return "", err
+	}
+	return etag, nil
+}
+
+// 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.
+func (up *multipartUpload) streamPart(partNumber int, size int64, md5Sum []byte, rw *pool.RW) error {
+	up.mu.Lock()
+	up.partMD5s[partNumber] = md5Sum
+	up.partSizes[partNumber] = size
+	up.streamBuf[partNumber] = rw
+
+	if up.pumping {
+		// Another goroutine owns the pipe and will pump this part in turn.
+		up.mu.Unlock()
+		return nil
+	}
+	up.pumping = true
+
+	for {
+		prw, ok := up.streamBuf[up.nextPart]
+		if !ok {
+			up.pumping = false
+			up.mu.Unlock()
+			return nil
+		}
+		delete(up.streamBuf, up.nextPart)
+		up.mu.Unlock()
+
+		err := pipePart(up.pipeW, prw)
+		_ = prw.Close()
+		if err != nil {
+			up.mu.Lock()
+			up.pumping = false
+			up.mu.Unlock()
+			return err
+		}
+
+		up.mu.Lock()
+		up.nextPart++
+	}
+}
+
+// pipePart writes the whole of rw into w (the pipe).
+func pipePart(w io.Writer, rw *pool.RW) error {
+	if _, err := rw.Seek(0, io.SeekStart); err != nil {
+		return err
+	}
+	_, err := io.Copy(w, rw)
+	return err
+}
+
+// CompleteMultipartUpload finalises a streamed multipart upload. It closes the
+// pipe (so PutStream finishes), registers the new file with the VFS, computes
+// the S3-style multipart ETag, and stores the user metadata so HeadObject and
+// GetObject see the same fields the in-memory PutObject path produces.
+func (b *s3Backend) CompleteMultipartUpload(ctx context.Context, bucketName, objectName string, uploadID gofakes3.UploadID, input *gofakes3.CompleteMultipartUploadRequest) (gofakes3.VersionID, string, error) {
+	up, err := b.loadUpload(uploadID)
+	if err != nil {
+		return "", "", err
+	}
+	defer b.multipartUploads.Delete(uploadID)
+
+	if err := up.validate(input); err != nil {
+		_ = up.abort(ctx)
+		return "", "", err
+	}
+
+	// All parts must have been streamed: contiguous part numbers from 1 with
+	// nothing left buffered. A leftover means the client used non-contiguous
+	// part numbers, which the in-order stream can't place.
+	up.mu.Lock()
+	streamed := up.nextPart - 1
+	total := len(up.partSizes)
+	leftover := len(up.streamBuf)
+	up.mu.Unlock()
+	if leftover != 0 || streamed != total {
+		_ = up.abort(ctx)
+		return "", "", gofakes3.ErrInvalidPart
+	}
+
+	if err := up.close(ctx); err != nil {
+		return "", "", err
+	}
+
+	_vfs, err := b.s.getVFS(ctx)
+	if err != nil {
+		return "", "", err
+	}
+
+	// Invalidate the parent directory's cached listing so subsequent VFS
+	// Stat / List calls pick up the newly-written object from the
+	// underlying Fs (we wrote to the Fs directly, bypassing VFS).
+	if root, err := _vfs.Root(); err == nil {
+		root.ForgetPath(up.fp, fs.EntryObject)
+	}
+
+	b.meta.Store(up.fp, up.meta)
+	if val, ok := up.meta["X-Amz-Meta-Mtime"]; ok {
+		if ti, err := swift.FloatStringToTime(val); err == nil {
+			b.storeModtime(up.fp, up.meta, val)
+			_ = _vfs.Chtimes(up.fp, ti, ti)
+		}
+	} else if val, ok := up.meta["mtime"]; ok {
+		if ti, err := swift.FloatStringToTime(val); err == nil {
+			b.storeModtime(up.fp, up.meta, val)
+			_ = _vfs.Chtimes(up.fp, ti, ti)
+		}
+	}
+
+	return "", up.multipartETag(input), nil
+}
+
+// AbortMultipartUpload tears down an in-progress upload, asking the background
+// PutStream to discard any data already sent.
+func (b *s3Backend) AbortMultipartUpload(ctx context.Context, bucketName, objectName string, uploadID gofakes3.UploadID) error {
+	up, err := b.loadUpload(uploadID)
+	if err != nil {
+		return err
+	}
+	defer b.multipartUploads.Delete(uploadID)
+	return up.abort(ctx)
+}
+
+// validate cross-checks the part list supplied by the client against the
+// parts we actually received.
+func (up *multipartUpload) validate(input *gofakes3.CompleteMultipartUploadRequest) error {
+	up.mu.Lock()
+	defer up.mu.Unlock()
+
+	for i := 1; i < len(input.Parts); i++ {
+		if input.Parts[i].PartNumber <= input.Parts[i-1].PartNumber {
+			return gofakes3.ErrInvalidPartOrder
+		}
+	}
+	if len(input.Parts) != len(up.partSizes) {
+		return gofakes3.ErrInvalidPart
+	}
+	for _, p := range input.Parts {
+		md5Sum, ok := up.partMD5s[p.PartNumber]
+		if !ok {
+			return gofakes3.ErrInvalidPart
+		}
+		clientETag := strings.Trim(p.ETag, `"`)
+		if clientETag != hex.EncodeToString(md5Sum) {
+			return gofakes3.ErrInvalidPart
+		}
+	}
+	return nil
+}
+
+// close finalises the upload by signalling EOF to the background PutStream and
+// waiting for it to finish.
+func (up *multipartUpload) close(ctx context.Context) error {
+	up.mu.Lock()
+	if up.closed {
+		up.mu.Unlock()
+		return nil
+	}
+	up.closed = true
+	up.mu.Unlock()
+
+	err := up.pipeW.Close()
+	<-up.putDone
+	up.putCancel()
+	if up.putErr != nil {
+		return up.putErr
+	}
+	return err
+}
+
+// errMultipartAborted makes the background PutStream fail when an upload is
+// aborted, so it tears down its partial object instead of completing.
+var errMultipartAborted = errors.New("serve s3: multipart upload aborted")
+
+// abort cancels the background PutStream and releases any buffered parts.
+func (up *multipartUpload) abort(ctx context.Context) error {
+	up.mu.Lock()
+	if up.closed {
+		up.mu.Unlock()
+		return nil
+	}
+	up.closed = true
+	streamBuf := up.streamBuf
+	up.streamBuf = nil
+	up.mu.Unlock()
+
+	for _, rw := range streamBuf {
+		_ = rw.Close()
+	}
+
+	// Fail the background PutStream (so it discards its partial object) and
+	// wait for it to return.
+	up.putCancel()
+	_ = up.pipeW.CloseWithError(errMultipartAborted)
+	<-up.putDone
+	return nil
+}
+
+// multipartETag computes the S3 multipart ETag for the assembled object:
+//
+//	hex(md5(concat(part_md5s_in_order))) + "-" + N
+func (up *multipartUpload) multipartETag(input *gofakes3.CompleteMultipartUploadRequest) string {
+	partNumbers := make([]int, 0, len(input.Parts))
+	for _, p := range input.Parts {
+		partNumbers = append(partNumbers, p.PartNumber)
+	}
+	sort.Ints(partNumbers)
+
+	up.mu.Lock()
+	concat := make([]byte, 0, len(partNumbers)*md5.Size)
+	for _, n := range partNumbers {
+		concat = append(concat, up.partMD5s[n]...)
+	}
+	up.mu.Unlock()
+
+	sum := md5.Sum(concat)
+	return fmt.Sprintf("%q", fmt.Sprintf("%s-%d", hex.EncodeToString(sum[:]), len(partNumbers)))
+}

cmd/serve/s3/multipart_test.go

@@ -0,0 +1,198 @@
+// Multipart upload tests for serve s3.
+
+package s3
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net/url"
+	"path"
+	"sync"
+	"testing"
+
+	"github.com/minio/minio-go/v7"
+	"github.com/minio/minio-go/v7/pkg/credentials"
+	"github.com/rclone/rclone/cmd/serve/proxy"
+	"github.com/rclone/rclone/fs"
+	"github.com/rclone/rclone/fstest"
+	"github.com/rclone/rclone/lib/random"
+	"github.com/rclone/rclone/vfs/vfscommon"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// newMultipartTestServer starts a serve s3 server backed by a fresh local temp
+// directory and returns a low-level minio Core client (for explicit control of
+// the multipart parts), the backing Fs and the bucket name. The server and
+// client are torn down via t.Cleanup.
+func newMultipartTestServer(t *testing.T, disableStreaming bool) (*minio.Core, fs.Fs, string) {
+	fstest.Initialise()
+	ctx := context.Background()
+	f, err := fs.NewFs(ctx, t.TempDir())
+	require.NoError(t, err)
+	const bucket = "test"
+	require.NoError(t, f.Mkdir(ctx, bucket))
+
+	keyid := random.String(16)
+	keysec := random.String(16)
+	opt := Opt
+	opt.DisableMultipartStreaming = disableStreaming
+	opt.AuthKey = []string{fmt.Sprintf("%s,%s", keyid, keysec)}
+	opt.HTTP.ListenAddr = []string{endpoint}
+	w, err := newServer(ctx, f, &opt, &vfscommon.Opt, &proxy.Opt)
+	require.NoError(t, err)
+	go func() { _ = w.Serve() }()
+	t.Cleanup(func() { _ = w.Shutdown() })
+
+	u, err := url.Parse(w.server.URLs()[0])
+	require.NoError(t, err)
+	core, err := minio.NewCore(u.Host, &minio.Options{
+		Creds:  credentials.NewStaticV4(keyid, keysec, ""),
+		Secure: false,
+	})
+	require.NoError(t, err)
+	return core, f, bucket
+}
+
+// readObject reads bucket/object back from the backing Fs.
+func readObject(t *testing.T, f fs.Fs, bucket, object string) []byte {
+	ctx := context.Background()
+	o, err := f.NewObject(ctx, path.Join(bucket, object))
+	require.NoError(t, err)
+	rc, err := o.Open(ctx)
+	require.NoError(t, err)
+	got, err := io.ReadAll(rc)
+	require.NoError(t, err)
+	require.NoError(t, rc.Close())
+	return got
+}
+
+// multipartUploadParts uploads object to bucket as a multipart upload with the
+// given (in-order) part sizes and returns the assembled contents plus the
+// first error encountered.
+func multipartUploadParts(t *testing.T, core *minio.Core, bucket, object string, partSizes []int) ([]byte, error) {
+	ctx := context.Background()
+	uploadID, err := core.NewMultipartUpload(ctx, bucket, object, minio.PutObjectOptions{})
+	if err != nil {
+		return nil, err
+	}
+	var want []byte
+	var parts []minio.CompletePart
+	for i, sz := range partSizes {
+		data := []byte(random.String(sz))
+		want = append(want, data...)
+		p, err := core.PutObjectPart(ctx, bucket, object, uploadID, i+1, bytes.NewReader(data), int64(sz), minio.PutObjectPartOptions{})
+		if err != nil {
+			_ = core.AbortMultipartUpload(ctx, bucket, object, uploadID)
+			return want, err
+		}
+		parts = append(parts, minio.CompletePart{PartNumber: i + 1, ETag: p.ETag})
+	}
+	_, err = core.CompleteMultipartUpload(ctx, bucket, object, uploadID, parts, minio.PutObjectOptions{})
+	return want, err
+}
+
+// TestMultipartNonUniform checks that a multipart upload whose parts are NOT a
+// uniform size round-trips correctly, both with the default streaming path and
+// with the in-memory fallback (--disable-multipart-streaming).
+func TestMultipartNonUniform(t *testing.T) {
+	// Non-uniform parts, last one smaller.
+	partSizes := []int{120 * 1024, 100 * 1024, 53 * 1024}
+	const object = "non-uniform.bin"
+
+	for _, tc := range []struct {
+		name             string
+		disableStreaming bool
+	}{
+		{"Streaming", false},
+		{"InMemory", true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			core, f, bucket := newMultipartTestServer(t, tc.disableStreaming)
+			want, err := multipartUploadParts(t, core, bucket, object, partSizes)
+			require.NoError(t, err)
+			assert.Equal(t, want, readObject(t, f, bucket, object))
+		})
+	}
+}
+
+// TestMultipartOutOfOrder uploads the parts concurrently and out of order,
+// exercising the reorder buffer and the in-order pump handoff.
+func TestMultipartOutOfOrder(t *testing.T) {
+	core, f, bucket := newMultipartTestServer(t, false)
+	ctx := context.Background()
+	const object = "out-of-order.bin"
+
+	sizes := []int{70 * 1024, 90 * 1024, 50 * 1024, 33 * 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)
+
+	parts := make([]minio.CompletePart, len(sizes))
+	errs := make([]error, len(sizes))
+	var wg sync.WaitGroup
+	for _, i := range []int{2, 0, 3, 1} { // shuffled upload order
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			p, err := core.PutObjectPart(ctx, bucket, object, uploadID, i+1, bytes.NewReader(datas[i]), int64(sizes[i]), minio.PutObjectPartOptions{})
+			errs[i] = err
+			parts[i] = minio.CompletePart{PartNumber: i + 1, ETag: p.ETag}
+		}(i)
+	}
+	wg.Wait()
+	for _, err := range errs {
+		require.NoError(t, err)
+	}
+
+	_, err = core.CompleteMultipartUpload(ctx, bucket, object, uploadID, parts, minio.PutObjectOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, want, readObject(t, f, bucket, object))
+}
+
+// TestMultipartNonContiguous checks that a multipart upload with a gap in the
+// part numbers (which the in-order stream can't place) is rejected.
+func TestMultipartNonContiguous(t *testing.T) {
+	core, _, bucket := newMultipartTestServer(t, false)
+	ctx := context.Background()
+	const object = "gap.bin"
+
+	uploadID, err := core.NewMultipartUpload(ctx, bucket, object, minio.PutObjectOptions{})
+	require.NoError(t, err)
+
+	var parts []minio.CompletePart
+	for _, pn := range []int{1, 2, 4} { // part 3 missing
+		data := []byte(random.String(40 * 1024))
+		p, err := core.PutObjectPart(ctx, bucket, object, uploadID, pn, bytes.NewReader(data), int64(len(data)), minio.PutObjectPartOptions{})
+		require.NoError(t, err)
+		parts = append(parts, minio.CompletePart{PartNumber: pn, ETag: p.ETag})
+	}
+	_, err = core.CompleteMultipartUpload(ctx, bucket, object, uploadID, parts, minio.PutObjectOptions{})
+	require.Error(t, err)
+}
+
+// TestMultipartAbort checks that aborting an upload tears down the streamed
+// PutStream so no object is left behind.
+func TestMultipartAbort(t *testing.T) {
+	core, f, bucket := newMultipartTestServer(t, false)
+	ctx := context.Background()
+	const object = "aborted.bin"
+
+	uploadID, err := core.NewMultipartUpload(ctx, bucket, object, minio.PutObjectOptions{})
+	require.NoError(t, err)
+	data := []byte(random.String(50 * 1024))
+	_, err = core.PutObjectPart(ctx, bucket, object, uploadID, 1, bytes.NewReader(data), int64(len(data)), minio.PutObjectPartOptions{})
+	require.NoError(t, err)
+	require.NoError(t, core.AbortMultipartUpload(ctx, bucket, object, uploadID))
+
+	_, err = f.NewObject(ctx, path.Join(bucket, object))
+	require.ErrorIs(t, err, fs.ErrorObjectNotFound)
+}

cmd/serve/s3/s3.go

@@ -37,6 +37,10 @@ var OptionsInfo = fs.Options{{
 	Name:    "no_cleanup",
 	Default: false,
 	Help:    "Not to cleanup empty folder after object is deleted",
+}, {
+	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)",
 }}.
 	Add(httplib.ConfigInfo).
 	Add(httplib.AuthConfigInfo)
@@ -44,12 +48,13 @@ 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"`
-	Auth           httplib.AuthConfig
+	DisableMultipartStreaming bool     `config:"disable_multipart_streaming"`
-	HTTP           httplib.Config
+	Auth                      httplib.AuthConfig
+	HTTP                      httplib.Config
 }
 
 // Opt is options set by command line flags

cmd/serve/s3/s3_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
 	_ "github.com/rclone/rclone/backend/local"
+	_ "github.com/rclone/rclone/backend/s3" // for TestS3Minio backing remote
 	"github.com/rclone/rclone/cmd/serve/proxy"
 	"github.com/rclone/rclone/cmd/serve/servetest"
 	"github.com/rclone/rclone/fs"
@@ -52,10 +53,10 @@ func serveS3(t *testing.T, f fs.Fs) (testURL string, keyid string, keysec string
 	return
 }
 
-// TestS3 runs the s3 server then runs the unit tests for the
+// startS3 builds the start callback that brings up a serve s3 server
-// s3 remote against it.
+// wrapping f and returns the client config for connecting to it.
-func TestS3(t *testing.T) {
+func startS3(t *testing.T) servetest.StartFn {
-	start := func(f fs.Fs) (configmap.Simple, func()) {
+	return func(f fs.Fs) (configmap.Simple, func()) {
 		testURL, keyid, keysec, _ := serveS3(t, f)
 		// Config for the backend we'll use to connect to the server
 		config := configmap.Simple{
@@ -65,11 +66,25 @@ func TestS3(t *testing.T) {
 			"access_key_id":     keyid,
 			"secret_access_key": keysec,
 		}
-
 		return config, func() {}
 	}
+}
 
-	servetest.Run(t, "s3", start)
+// TestS3 runs the s3 server backed by a local directory then runs the
+// s3 backend integration tests against it. The local backend only
+// supports OpenWriterAt, so this exercises that streaming path in
+// serve s3.
+func TestS3(t *testing.T) {
+	servetest.Run(t, "s3", startS3(t))
+}
+
+// TestS3Minio runs the s3 server backed by a minio docker container
+// (via fstest/testserver/init.d/TestS3Minio) then runs the s3 backend
+// integration tests against it. Minio supports OpenChunkWriter, so
+// this exercises that streaming path in serve s3 - the path the local
+// backing in TestS3 cannot cover.
+func TestS3Minio(t *testing.T) {
+	servetest.RunWithBackend(t, "s3", startS3(t), "TestS3Minio:")
 }
 
 // tests using the minio client

cmd/serve/s3/serve_s3.md

@@ -86,19 +86,65 @@ provider = Rclone
 endpoint = http://127.0.0.1:8080/
 access_key_id = ACCESS_KEY_ID
 secret_access_key = SECRET_ACCESS_KEY
-use_multipart_uploads = false
 ```
 
-Note that setting `use_multipart_uploads = false` is to work around
+### Multipart uploads
-[a bug](#bugs) which will be fixed in due course.
+
+By default `serve s3` **streams** each multipart upload, in part-number
+order, into a single `PutStream` upload to the underlying remote, so the
+whole file is never buffered in memory - memory use stays bounded by the
+parts in flight. The remote then performs its own internal upload (for
+example its own multipart upload, still with bounded memory). This works
+for any remote that supports `PutStream`, which is nearly all of them,
+including through `crypt`.
+
+**Advantages**
+
+- The whole object is never buffered in memory; memory use is bounded by
+  the parts in flight, not the upload size.
+- Parts can be any size. Clients that don't produce uniform-sized parts
+  work fine - for example PostgreSQL backup tools such as **pgBarman**
+  and **pgBackRest**, which flush an upload buffer once it grows past
+  the chunk size, so each part is the chunk size plus a variable
+  overshoot.
+- Works through `crypt` for any part size, since the object is encrypted
+  as one continuous stream.
+- Backend-agnostic - it only needs the remote to support `PutStream`.
+
+**Limitations**
+
+- 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.)
+- Parts are serialised into one stream, so ingest from the client is
+  effectively single-threaded, although the remote's own upload still
+  runs concurrently.
+
+#### Disabling streaming
+
+If you pass `--disable-multipart-streaming`, or the remote doesn't
+support `PutStream`, multipart uploads are instead **buffered in memory**
+by the underlying S3 library: every part is held in memory and the whole
+object is written out in one go when the upload completes (the previous
+behaviour). This removes the in-order/contiguous-part restriction above,
+so parts can be uploaded in any order, but **memory use grows with the
+size of the upload**, so it is only suitable for small objects. A one-off
+`NOTICE` is logged the first time this happens.
+
+Alternatively, if the client is an rclone `s3` remote (like the
+`[serves3]` example above), you can set `use_multipart_uploads = false`
+on it so it uploads each object as a single stream and skips multipart
+uploads altogether.
 
 ### Bugs
 
-When uploading multipart files `serve s3` holds all the parts in
-memory (see [#7453](https://github.com/rclone/rclone/issues/7453)).
-This is a limitaton of the library rclone uses for serving S3 and will
-hopefully be fixed at some point.
-
 Multipart server side copies do not work (see
 [#7454](https://github.com/rclone/rclone/issues/7454)). These take a
 very long time and eventually fail. The default threshold for

go.mod

@@ -73,7 +73,7 @@ require (
 	github.com/quasilyte/go-ruleguard/dsl v0.3.23
 	github.com/rclone/Proton-API-Bridge v1.0.3
 	github.com/rclone/go-proton-api v1.0.2
-	github.com/rclone/gofakes3 v0.0.6
+	github.com/rclone/gofakes3 v0.0.7
 	github.com/rfjakob/eme v1.2.0
 	github.com/rivo/uniseg v0.4.7
 	github.com/rogpeppe/go-internal v1.14.1

go.sum

@@ -598,8 +598,8 @@ github.com/rclone/Proton-API-Bridge v1.0.3 h1:Bs7RC4xCFSN0BPIYVda/BNxp0qo3NV0gB2
 github.com/rclone/Proton-API-Bridge v1.0.3/go.mod h1:26RAest751Ofk+F/d8xtl4UyWXrZvMQwn39U8rm/WKM=
 github.com/rclone/go-proton-api v1.0.2 h1:cJtJUab0MGJ3C6q5kiEJs3pbyhSLnOKMyYOQehA0PBc=
 github.com/rclone/go-proton-api v1.0.2/go.mod h1:LB2kCEaZMzNn3ocdz+qYfxXmuLxxN0ka62KJd2x53Bc=
-github.com/rclone/gofakes3 v0.0.6 h1:gTW5Dq04EouaoKyyTSuSL+WS3VZU3k2QNOBn9ghX7kE=
+github.com/rclone/gofakes3 v0.0.7 h1:5AeROIAFbhp/7+f7BSsc8LzDbnx/WxgjV22SAUKmHp4=
-github.com/rclone/gofakes3 v0.0.6/go.mod h1:M/uFc52GcEFuIP4stgrSR99SqqqxMR0ncgiTbj0eszc=
+github.com/rclone/gofakes3 v0.0.7/go.mod h1:M/uFc52GcEFuIP4stgrSR99SqqqxMR0ncgiTbj0eszc=
 github.com/relvacode/iso8601 v1.7.0 h1:BXy+V60stMP6cpswc+a93Mq3e65PfXCgDFfhvNNGrdo=
 github.com/relvacode/iso8601 v1.7.0/go.mod h1:FlNp+jz+TXpyRqgmM7tnzHHzBnz776kmAH2h3sZCn0I=
 github.com/rfjakob/eme v1.2.0 h1:8dAHL+WVAw06+7DkRKnRiFp1JL3QjcJEZFqDnndUaSI=