diff --git a/cmd/serve/s3/list.go b/cmd/serve/s3/list.go index 3f87f7ccb..76908e8c0 100644 --- a/cmd/serve/s3/list.go +++ b/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) diff --git a/cmd/serve/s3/multipart.go b/cmd/serve/s3/multipart.go index ab589173e..f5c7d56cc 100644 --- a/cmd/serve/s3/multipart.go +++ b/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 { diff --git a/cmd/serve/s3/multipart_test.go b/cmd/serve/s3/multipart_test.go index 48b907166..4aa812250 100644 --- a/cmd/serve/s3/multipart_test.go +++ b/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) + }) + } } diff --git a/cmd/serve/s3/serve_s3.md b/cmd/serve/s3/serve_s3.md index 949a21e60..d4457cbb2 100644 --- a/cmd/serve/s3/serve_s3.md +++ b/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,