serve s3: fix streamed multipart uploads not being atomic

Before this change a streamed multipart upload wrote its parts
straight to the object's final path on the underlying remote. That
meant an in-progress upload overwrote any object already stored under
that name, and aborting or failing the upload destroyed it. The
opposite of the S3 guarantee that an object only changes on a
successful CompleteMultipartUpload.

Remotes that upload atomically already (PartialUploads is false, e.g.
object stores) are safe to stream straight to the destination, so they
still do. Remotes where a partial upload is visible (PartialUploads is
true, e.g. local) now stream the parts to a temporary object instead
and move it, server-side, into its final place only when the upload
completes. A failed or aborted upload then just removes the temporary
object and leaves any pre-existing object untouched.

The temporary-object path needs the remote to support a server-side
move or copy in addition to PutStream uploads fall back to being
buffered in memory as before. The temporary objects are named with a
leading ".rclone_multipart_upload_" and hidden from listings.

cmd/serve/s3/list.go

@@ -23,6 +23,11 @@ func (b *s3Backend) entryListR(_vfs *vfs.VFS, bucketName, fdPath, name string, a
 	for _, entry := range dirEntries {
 		object := entry.Name()
 
+		// Hide the in-progress multipart uploads
+		if strings.HasPrefix(object, multipartUploadPrefix) {
+			continue
+		}
+
 		// workaround for control-chars detect
 		objectPath := path.Join(fdPath, object)
 

cmd/serve/s3/multipart.go

@@ -29,15 +29,21 @@ import (
 	"github.com/rclone/gofakes3"
 	"github.com/rclone/rclone/fs"
 	"github.com/rclone/rclone/fs/object"
+	"github.com/rclone/rclone/fs/operations"
 	"github.com/rclone/rclone/lib/multipart"
 	"github.com/rclone/rclone/lib/pool"
 )
 
+// multipartUploadPrefix is prepended to the leaf name of the temporary object
+// a streamed multipart upload is written to before it is moved into place.
+const multipartUploadPrefix = ".rclone_multipart_upload_"
+
 // 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)
+	fp          string // final object path
+	streamFp    string // path the parts are streamed to (fp when the backend uploads atomically)
 	meta        map[string]string
 
 	pipeW *io.PipeWriter // parts are streamed here, in part-number order
@@ -56,11 +62,12 @@ type multipartUpload struct {
 }
 
 // newMultipartUpload allocates an upload struct.
-func newMultipartUpload(bucket, key, fp string, meta map[string]string) *multipartUpload {
+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{},
@@ -81,10 +88,19 @@ func (b *s3Backend) loadUpload(uploadID gofakes3.UploadID) (*multipartUpload, er
 // 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.
+// Backends that upload atomically (PartialUploads=false) are streamed
+// straight to the final object. An aborted or failed upload never
+// makes a partial object visible or disturbs a pre-existing one.
+// Backends where a partial upload is visible (PartialUploads=true)
+// are instead streamed to a temporary object that is moved into
+// place, server-side, on completion, giving the same atomic
+// behaviour.
+//
+// If streaming is disabled (--disable-multipart-streaming), the Fs has no
+// PutStream, or a non-atomic Fs can't move/copy objects server-side,
+// 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 {
@@ -96,12 +112,8 @@ func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objec
 
 	f := _vfs.Fs()
 	features := f.Features()
-	if b.s.opt.DisableMultipartStreaming || features.PutStream == nil {
+	if reason := b.noStreamingReason(f); reason != "" {
 		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
@@ -118,9 +130,16 @@ func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objec
 		}
 	}
 
-	up := newMultipartUpload(bucketName, objectName, fp, meta)
+	uploadID := gofakes3.UploadID(uuid.New().String())
+	streamFp := fp
+	// If partial uploads visible, stream to temporary object
+	if features.PartialUploads {
+		streamFp = path.Join(objectDir, multipartUploadPrefix+string(uploadID))
+	}
 
-	src := object.NewStaticObjectInfo(fp, time.Now(), -1, true, nil, f)
+	up := newMultipartUpload(bucketName, objectName, fp, streamFp, meta)
+
+	src := object.NewStaticObjectInfo(streamFp, 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.
@@ -135,11 +154,25 @@ func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objec
 		close(up.putDone)
 	}()
 
-	uploadID := gofakes3.UploadID(uuid.New().String())
 	b.multipartUploads.Store(uploadID, up)
 	return uploadID, nil
 }
 
+// noStreamingReason returns a non-empty reason why streamed multipart uploads
+// can't be used for f, or "" if they can.
+func (b *s3Backend) noStreamingReason(f fs.Fs) string {
+	switch {
+	case b.s.opt.DisableMultipartStreaming:
+		return "--disable-multipart-streaming is set"
+	case f.Features().PutStream == nil:
+		return "this backend doesn't support streaming uploads"
+	case f.Features().PartialUploads && !operations.CanServerSideMove(f):
+		return "this backend can't upload atomically and has no server-side move or copy"
+	default:
+		return ""
+	}
+}
+
 // 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)
@@ -235,7 +268,7 @@ func (b *s3Backend) CompleteMultipartUpload(ctx context.Context, bucketName, obj
 
 	if err := up.validate(input); err != nil {
 		_ = up.abort(ctx)
-		b.forgetPath(ctx, up.fp)
+		b.forgetPath(ctx, up.streamFp)
 		return "", "", err
 	}
 
@@ -249,11 +282,12 @@ func (b *s3Backend) CompleteMultipartUpload(ctx context.Context, bucketName, obj
 	up.mu.Unlock()
 	if leftover != 0 || streamed != total {
 		_ = up.abort(ctx)
-		b.forgetPath(ctx, up.fp)
+		b.forgetPath(ctx, up.streamFp)
 		return "", "", gofakes3.ErrInvalidPart
 	}
 
 	if err := up.close(ctx); err != nil {
+		b.forgetPath(ctx, up.streamFp)
 		return "", "", err
 	}
 
@@ -262,6 +296,15 @@ func (b *s3Backend) CompleteMultipartUpload(ctx context.Context, bucketName, obj
 		return "", "", err
 	}
 
+	// If the parts were streamed to a temporary object move it into place
+	if up.streamFp != up.fp {
+		if err := b.moveIntoPlace(ctx, _vfs.Fs(), up.streamFp, up.fp); err != nil {
+			b.forgetPath(ctx, up.streamFp)
+			b.forgetPath(ctx, up.fp)
+			return "", "", err
+		}
+		b.forgetPath(ctx, up.streamFp)
+	}
 	b.forgetPath(ctx, up.fp)
 
 	b.meta.Store(up.fp, up.meta)
@@ -289,16 +332,28 @@ func (b *s3Backend) AbortMultipartUpload(ctx context.Context, bucketName, object
 	}
 	defer b.multipartUploads.Delete(uploadID)
 	err = up.abort(ctx)
-	b.forgetPath(ctx, up.fp)
+	// An atomic backend leaves the final object untouched by the aborted
+	// upload; a non-atomic one only ever wrote the temporary object. Either way
+	// invalidating streamFp is enough.
+	b.forgetPath(ctx, up.streamFp)
 	return err
 }
 
+// moveIntoPlace moves the temporary object srcFp to its final path dstFp on f,
+// server-side, overwriting any object already there.
+func (b *s3Backend) moveIntoPlace(ctx context.Context, f fs.Fs, srcFp, dstFp string) error {
+	srcObj, err := f.NewObject(ctx, srcFp)
+	if err != nil {
+		return fmt.Errorf("failed to find uploaded object: %w", err)
+	}
+	if _, err := operations.Move(ctx, f, nil, dstFp, srcObj); err != nil {
+		return fmt.Errorf("failed to move uploaded object into place: %w", err)
+	}
+	return nil
+}
+
 // forgetPath invalidates the parent directory's cached VFS listing so that
-// subsequent VFS Stat / List calls re-read up.fp from the underlying Fs. The
-// streamed multipart path writes to (and, on abort, removes from) the Fs
-// directly, bypassing the VFS, so the VFS cache would otherwise keep serving a
-// stale entry - including a ghost of a pre-existing object that an aborted
-// upload has overwritten and removed.
+// subsequent VFS Stat / List calls re-read fp from the underlying Fs.
 func (b *s3Backend) forgetPath(ctx context.Context, fp string) {
 	_vfs, err := b.s.getVFS(ctx)
 	if err != nil {

cmd/serve/s3/multipart_test.go

@@ -10,30 +10,56 @@ import (
 	"net/url"
 	"path"
 	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
+	_ "github.com/rclone/rclone/backend/memory"
 	"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"
 	"github.com/rclone/rclone/vfs/vfscommon"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+// testBackingCounter hands out unique backing roots across test servers.
+var testBackingCounter atomic.Int64
+
 // 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) {
+	return newMultipartTestServerBacking(t, "", disableStreaming)
+}
+
+// newMultipartTestServerBacking is like newMultipartTestServer but backed by
+// the named remote (a fresh local temp directory if empty). ":memory:" gives
+// an atomic (PartialUploads=false) backing, so the streamed-straight-to-the-
+// 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) {
 	fstest.Initialise()
 	ctx := context.Background()
-	f, err := fs.NewFs(ctx, t.TempDir())
+	if backing == "" {
+		backing = t.TempDir()
+	}
+	f, err := fs.NewFs(ctx, backing)
 	require.NoError(t, err)
-	const bucket = "test"
+	// A unique bucket per server: every plain ":memory:" backing shares one
+	// process-wide store, so a fixed name would leak objects between tests.
+	bucket := fmt.Sprintf("test-%d", testBackingCounter.Add(1))
 	require.NoError(t, f.Mkdir(ctx, bucket))
+	// The VFS is cached per remote (fs.ConfigString), so a shared ":memory:"
+	// server reuses a VFS whose cached root listing predates the bucket just
+	// created; forget it so the new bucket is visible.
+	if root, err := vfs.New(ctx, f, &vfscommon.Opt).Root(); err == nil {
+		root.ForgetAll()
+	}
 
 	keyid := random.String(16)
 	keysec := random.String(16)
@@ -179,20 +205,98 @@ func TestMultipartNonContiguous(t *testing.T) {
 	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{})
+// requireOnly asserts that the bucket contains only the expected
+// objects, in particular no leftover temporary multipart objects.
+func requireOnly(t *testing.T, f fs.Fs, bucket string, want ...string) {
+	entries, err := f.List(context.Background(), bucket)
 	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))
+	var got []string
+	for _, entry := range entries {
+		got = append(got, path.Base(entry.Remote()))
+	}
+	assert.ElementsMatch(t, want, got)
+}
 
-	_, err = f.NewObject(ctx, path.Join(bucket, object))
-	require.ErrorIs(t, err, fs.ErrorObjectNotFound)
+// testRemotes to exercise all the code branches
+var testRemotes = []struct {
+	name    string
+	backing string
+}{
+	{"Local", ""},          // PartialUploads=true
+	{"Memory", ":memory:"}, // PartialUploads=false
+}
+
+// TestMultipartAbort checks that aborting an upload tears down the streamed
+// PutStream so neither the object nor its temporary object is left behind.
+func TestMultipartAbort(t *testing.T) {
+	for _, tc := range testRemotes {
+		t.Run(tc.name, func(t *testing.T) {
+			core, f, bucket := newMultipartTestServerBacking(t, tc.backing, 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)
+			requireOnly(t, f, bucket)
+		})
+	}
+}
+
+// TestMultipartAbortPreservesExisting checks that aborting an upload to a name
+// that already holds an object leaves the existing object untouched - the
+// streamed upload must be atomic, not overwrite the destination as it goes.
+func TestMultipartAbortPreservesExisting(t *testing.T) {
+	for _, tc := range testRemotes {
+		t.Run(tc.name, func(t *testing.T) {
+			core, f, bucket := newMultipartTestServerBacking(t, tc.backing, false)
+			ctx := context.Background()
+			const object = "existing.bin"
+
+			// Put an object the normal (non-multipart) way.
+			existing := []byte(random.String(100))
+			_, err := core.PutObject(ctx, bucket, object, bytes.NewReader(existing), int64(len(existing)), "", "", minio.PutObjectOptions{})
+			require.NoError(t, err)
+
+			// Start a multipart upload to the same name, upload a part, then abort.
+			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))
+
+			// The original object must survive, and no temporary object be left behind.
+			assert.Equal(t, existing, readObject(t, f, bucket, object))
+			requireOnly(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) {
+	for _, tc := range testRemotes {
+		t.Run(tc.name, func(t *testing.T) {
+			core, f, bucket := newMultipartTestServerBacking(t, tc.backing, false)
+			ctx := context.Background()
+			const object = "overwrite.bin"
+
+			existing := []byte(random.String(100))
+			_, err := core.PutObject(ctx, bucket, object, bytes.NewReader(existing), int64(len(existing)), "", "", minio.PutObjectOptions{})
+			require.NoError(t, err)
+
+			want, err := multipartUploadParts(t, core, bucket, object, []int{60 * 1024, 40 * 1024})
+			require.NoError(t, err)
+
+			assert.Equal(t, want, readObject(t, f, bucket, object))
+			requireOnly(t, f, bucket, object)
+		})
+	}
 }

cmd/serve/s3/serve_s3.md

@@ -98,7 +98,16 @@ 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 upload is atomic so the destination object only ever changes on a
+successful completion. A failed or aborted upload never affects any
+object already stored under that name. Remotes that upload atomically
+already (object stores such as `s3`) are streamed straight to the
+destination. On remotes where a partial upload would otherwise be visible
+(such as `local`), the parts are streamed to a temporary object that is
+moved into place, server-side, on completion; these remotes therefore
+also need to support a server-side move or copy.
+
+**Features**
 
 - The whole object is never buffered in memory; memory use is bounded by
   the parts in flight, not the upload size.
@@ -109,7 +118,12 @@ including through `crypt`.
   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`.
+- The destination object only ever changes atomically, on completion: an
+  aborted or failed upload leaves any pre-existing object of the same
+  name untouched, and a partly-uploaded object never becomes visible.
+- Backend-agnostic - it only needs the remote to support `PutStream`
+  (plus a server-side move or copy on remotes that don't upload
+  atomically).
 
 **Limitations**
 
@@ -126,11 +140,18 @@ including through `crypt`.
 - Parts are serialised into one stream, so ingest from the client is
   effectively single-threaded, although the remote's own upload still
   runs concurrently.
+- On remotes that don't upload atomically (such as `local`), the
+  completed object is moved into place with a server-side operation.
+  This is a cheap rename on most such remotes. On these remotes, if
+  `serve s3` is killed part-way through an upload the temporary object
+  (named with a leading `.rclone_multipart_upload_`) may be left behind;
+  it is hidden from S3 listings but must be removed manually.
 
 #### Disabling streaming
 
 If you pass `--disable-multipart-streaming`, or the remote doesn't
-support `PutStream`, multipart uploads are instead **buffered in memory**
+support `PutStream` (or doesn't upload atomically and can't move or copy
+server-side), 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,