diff --git a/cmd/serve/s3/backend.go b/cmd/serve/s3/backend.go index afa4e2cae..8c135e969 100644 --- a/cmd/serve/s3/backend.go +++ b/cmd/serve/s3/backend.go @@ -371,10 +371,10 @@ func (b *s3Backend) PutObject( // (never the object at fp) and any stale VFS state for fp. cleanup := func() { if tmpFp != fp { - b.forgetPath(ctx, tmpFp) + b.forgetPath(_vfs, tmpFp) _ = _vfs.Remove(tmpFp) } else { - b.forgetPath(ctx, fp) + b.forgetPath(_vfs, fp) } } diff --git a/cmd/serve/s3/multipart.go b/cmd/serve/s3/multipart.go index dd38e3c3d..304aaf69f 100644 --- a/cmd/serve/s3/multipart.go +++ b/cmd/serve/s3/multipart.go @@ -1,13 +1,17 @@ // 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 +// Multipart uploads received by serve s3 are written, in part-number order, +// through the VFS, exactly like a plain PutObject: to a temporary object +// which is renamed into place on completion, so the object at the key only +// ever changes atomically on success. With the default --vfs-cache-mode off +// the parts stream through the VFS into a single upload to the remote; with +// --vfs-cache-mode writes or above they are buffered in the VFS cache and +// uploaded by its write-back. 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. +// When streaming is disabled (--disable-multipart-streaming), +// ErrMultipartUploadNotSupported is returned so that gofakes3 falls back to +// buffering the parts in memory. package s3 @@ -23,46 +27,47 @@ import ( "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/fs/operations" "github.com/rclone/rclone/lib/multipart" "github.com/rclone/rclone/lib/pool" + "github.com/rclone/rclone/vfs" + "github.com/rclone/rclone/vfs/vfscommon" ) // 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 = tempObjectPrefix + "multipart_" -// multipartUpload tracks one in-flight S3 multipart upload that is being -// streamed, in part order, into a single PutStream upload to the underlying Fs. +// multipartUpload tracks one in-flight S3 multipart upload. The parts are +// written, in part-number order, into fh - a VFS file handle which either +// streams straight through to the remote (the default) or is backed by the +// VFS cache (when the VFS is caching writes). type multipartUpload struct { bucket, key string fp string // final object path - streamFp string // path the parts are streamed to (fp when the backend uploads atomically) + streamFp string // path the parts are written to (fp when the remote has no server-side move or copy) meta map[string]string - pipeW *io.PipeWriter // parts are streamed here, in part-number order + fh io.WriteCloser // sink the in-order parts are written to + vfs *vfs.VFS // the VFS fh was created on, used for all later operations 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 + aborted bool // closed by abort, so nothing was committed - 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) + 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 sink + 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. @@ -92,22 +97,26 @@ func (b *s3Backend) loadUpload(uploadID gofakes3.UploadID) (*multipartUpload, er 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. +// CreateMultipartUpload begins a new multipart upload. // -// 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. +// The parts are written, in part-number order, through the VFS to a temporary +// object which is renamed into place on completion. With the default +// --vfs-cache-mode off the write streams through to the remote as the parts +// arrive; with --vfs-cache-mode writes or above it lands in the VFS cache and +// is uploaded by the write-back. Either way an aborted or failed upload never +// makes a partial object visible at the final path or disturbs a pre-existing +// one. // -// 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. +// On a remote with no server-side move or copy the parts are written straight +// to the final object instead, trading some atomicity for never buffering in +// memory: the in-flight upload is visible at the key, and a failed or aborted +// upload can leave partial data there when the remote doesn't upload atomically +// or the VFS is caching writes (a write to the cache can't be abandoned). +// +// With --disable-multipart-streaming, ErrMultipartUploadNotSupported is +// returned so that gofakes3 falls back to buffering the whole upload in +// memory; a one-off NOTICE warns about the memory use. A caching VFS needs +// no streaming support, so it ignores the flag. 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 { @@ -117,11 +126,9 @@ func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objec return "", gofakes3.BucketNotFound(bucketName) } - f := _vfs.Fs() - features := f.Features() - if reason := b.noStreamingReason(f); reason != "" { + if b.s.opt.DisableMultipartStreaming && _vfs.Opt.CacheMode < vfscommon.CacheModeMinimal { b.warnInMemoryOnce.Do(func() { - fs.Logf(nil, "serve s3: buffering multipart uploads in memory because %s - this may use a lot of memory", reason) + fs.Logf(nil, "serve s3: buffering multipart uploads in memory because --disable-multipart-streaming is set - this may use a lot of memory") }) return "", gofakes3.ErrMultipartUploadNotSupported } @@ -139,47 +146,28 @@ func (b *s3Backend) CreateMultipartUpload(ctx context.Context, bucketName, objec uploadID := gofakes3.UploadID(uuid.New().String()) streamFp := fp - // If partial uploads visible, stream to temporary object - if features.PartialUploads { + // Write to a temporary object moved into place on completion if the + // remote supports server-side move (if not write directly to the final + // object). Unlike a plain PutObject all remotes use a temporary object. + // S3 semantics say the key must not show it until it completes. This is + // at the cost of a server-side copy and delete where the remote has no + // server side move. + if operations.CanServerSideMove(_vfs.Fs()) { streamFp = path.Join(objectDir, multipartUploadPrefix+string(uploadID)) } 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() - // 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) - }() + fh, err := _vfs.Create(streamFp) + if err != nil { + return "", err + } + up.fh = fh + up.vfs = _vfs 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) @@ -222,7 +210,7 @@ func (b *s3Backend) UploadPart(ctx context.Context, bucketName, objectName strin // 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 +// admitted so the sink 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 { @@ -249,11 +237,11 @@ func (up *multipartUpload) release(size int64) { up.mu.Unlock() } -// streamPart records a part and streams the parts into the pipe in order. +// streamPart records a part and streams the parts into the sink 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 +// parts are then pumped into the sink in order. Whichever goroutine finds the // next part available does the pumping, so concurrent (but in-order) clients // are tolerated, with the buffering bounded by waitForTurn. // @@ -298,13 +286,21 @@ func (up *multipartUpload) streamPart(partNumber int, size int64, md5Sum []byte, up.streamBuf[partNumber] = rw if up.pumping { - // Another goroutine owns the pipe and will pump this part in turn. + // Another goroutine owns the sink and will pump this part in turn. up.mu.Unlock() return nil } up.pumping = true for { + if up.closed { + // Aborted while pumping: the sink is closed and any + // parts still buffered were released by the abort, so + // report the upload gone rather than success. + up.pumping = false + up.mu.Unlock() + return gofakes3.ErrNoSuchUpload + } prw, ok := up.streamBuf[up.nextPart] if !ok { up.pumping = false @@ -315,14 +311,21 @@ func (up *multipartUpload) streamPart(partNumber int, size int64, md5Sum []byte, psize := prw.Size() up.mu.Unlock() - err := pipePart(up.pipeW, prw) + err := pipePart(up.fh, prw) _ = prw.Close() if err != nil { up.mu.Lock() up.pumping = false up.buffered -= psize + closed := up.closed up.cond.Broadcast() up.mu.Unlock() + if closed { + // The write failed because the upload was aborted while + // this part was being pumped: report the upload gone + // rather than the sink's ECLOSED as an internal error. + return gofakes3.ErrNoSuchUpload + } return err } @@ -333,7 +336,7 @@ func (up *multipartUpload) streamPart(partNumber int, size int64, md5Sum []byte, } } -// pipePart writes the whole of rw into w (the pipe). +// pipePart writes the whole of rw into w (the sink). func pipePart(w io.Writer, rw *pool.RW) error { if _, err := rw.Seek(0, io.SeekStart); err != nil { return err @@ -342,110 +345,95 @@ func pipePart(w io.Writer, rw *pool.RW) error { 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 +// CompleteMultipartUpload finalises a multipart upload. It closes the sink, +// committing the upload, renames the temporary object into place, 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) - b.forgetPath(ctx, up.streamFp) + b.multipartUploads.Delete(uploadID) + _ = up.abort() + b.discardUpload(up) 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) - b.forgetPath(ctx, up.streamFp) - return "", "", gofakes3.ErrInvalidPart - } - - if err := up.close(ctx); err != nil { - b.forgetPath(ctx, up.streamFp) + // close commits the upload, failing with the upload left open if the + // streamed parts don't form the complete object; abort then tears it + // down. (After a successful or failed commit the abort is a no-op.) + if err := up.close(); err != nil { + b.multipartUploads.Delete(uploadID) + _ = up.abort() + b.discardUpload(up) return "", "", err } - _vfs, err := b.s.getVFS(ctx) - if err != nil { - return "", "", err - } - - // If the parts were streamed to a temporary object move it into place + // Rename the temporary object into place: on a caching VFS the + // write-back then uploads it under the final name, otherwise it is + // moved server-side on the remote. On failure the upload record is + // kept, because gofakes3 keeps its own record when the backend errors + // so that the client can retry the CompleteMultipartUpload: the + // committed close is idempotent, so the retry just renames again. 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) + if err := up.vfs.Rename(up.streamFp, up.fp); err != nil { return "", "", err } - b.forgetPath(ctx, up.streamFp) } - b.forgetPath(ctx, up.fp) + b.multipartUploads.Delete(uploadID) 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) + _ = up.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) + _ = up.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. +// AbortMultipartUpload tears down an in-progress upload, discarding any data +// already received. 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) - err = up.abort(ctx) - // 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 + if err := up.abort(); err != nil { + fs.Errorf(up.fp, "aborting multipart upload: %v", err) + } + b.discardUpload(up) + return nil } -// 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) +// discardUpload cleans up after a failed or aborted upload, removing the +// temporary object and any stale VFS state for it. It never removes the +// object at the final path: when the parts were written straight to the +// final object (a remote with no server-side move or copy) it may hold a +// pre-existing object the abandoned streaming write never disturbed, so +// only the VFS's view of the path is refreshed. (A caching VFS on such a +// remote commits the partial data instead - see abort.) +func (b *s3Backend) discardUpload(up *multipartUpload) { + b.forgetPath(up.vfs, up.streamFp) + if up.streamFp == up.fp { + return } - 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 + _ = up.vfs.Remove(up.streamFp) } // forgetPath invalidates the parent directory's cached VFS listing so that // 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 { - return - } +func (b *s3Backend) forgetPath(_vfs *vfs.VFS, fp string) { if root, err := _vfs.Root(); err == nil { root.ForgetPath(fp, fs.EntryObject) } @@ -478,39 +466,59 @@ func (up *multipartUpload) validate(input *gofakes3.CompleteMultipartUploadReque 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 { +// close finalises the upload, committing it: closing the sink completes the +// streaming upload to the remote, or the write to the VFS cache whose +// write-back then uploads it. +// +// The upload is only committed if the streamed parts form the complete +// object: 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). Otherwise ErrInvalidPart is returned and +// the upload is left open. The check and the commit share one critical +// section so no late part can slip into the stream between them. +// +// Returns ErrNoSuchUpload if the upload was aborted by a concurrent +// AbortMultipartUpload, in which case nothing has been committed - the +// upload must not be reported as complete. +func (up *multipartUpload) close() error { up.mu.Lock() if up.closed { + aborted := up.aborted up.mu.Unlock() + if aborted { + return gofakes3.ErrNoSuchUpload + } return nil } + if len(up.streamBuf) != 0 || up.nextPart-1 != len(up.partSizes) { + up.mu.Unlock() + return gofakes3.ErrInvalidPart + } up.closed = true up.cond.Broadcast() up.mu.Unlock() - err := up.pipeW.Close() - <-up.putDone - up.putCancel() - if up.putErr != nil { - return up.putErr - } - return err + return up.fh.Close() } -// errMultipartAborted makes the background PutStream fail when an upload is -// aborted, so it tears down its partial object instead of completing. +// errMultipartAborted is the reason an aborted upload's write is abandoned +// with, so the streaming upload fails instead of committing what it has. 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 { +// abort tears down the upload and releases any buffered parts. The write is +// abandoned so nothing is committed; a sink which can't abandon (a caching +// one) is closed normally, committing what it has to the cache - the caller +// then removes its temporary file, except on a remote with no server-side +// move or copy, where the parts went straight to the final path and the +// partial data is left to be written back. +func (up *multipartUpload) abort() error { up.mu.Lock() if up.closed { up.mu.Unlock() return nil } up.closed = true + up.aborted = true streamBuf := up.streamBuf up.streamBuf = nil up.cond.Broadcast() @@ -520,12 +528,11 @@ func (up *multipartUpload) abort(ctx context.Context) error { _ = 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 + if aborter, ok := up.fh.(interface{ CloseWithError(error) error }); ok { + _ = aborter.CloseWithError(errMultipartAborted) + return nil + } + return up.fh.Close() } // multipartETag computes the S3 multipart ETag for the assembled object: diff --git a/cmd/serve/s3/multipart_test.go b/cmd/serve/s3/multipart_test.go index 67c3a69bb..a0b585b56 100644 --- a/cmd/serve/s3/multipart_test.go +++ b/cmd/serve/s3/multipart_test.go @@ -22,6 +22,7 @@ import ( "github.com/rclone/rclone/cmd/serve/proxy" "github.com/rclone/rclone/fs" "github.com/rclone/rclone/fs/object" + "github.com/rclone/rclone/fs/operations" "github.com/rclone/rclone/fstest" "github.com/rclone/rclone/lib/multipart" "github.com/rclone/rclone/lib/random" @@ -44,9 +45,8 @@ func newMultipartTestServer(t *testing.T, disableStreaming bool) (*minio.Core, f // 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. +// an atomic (PartialUploads=false) backing, so both flavours of remote go +// through the temporary-object-plus-rename path. func newMultipartTestServerBacking(t *testing.T, backing string, disableStreaming bool) (*minio.Core, fs.Fs, string) { return newMultipartTestServerOpt(t, backing, disableStreaming, nil) } @@ -54,8 +54,22 @@ func newMultipartTestServerBacking(t *testing.T, backing string, disableStreamin // 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) { + return newMultipartTestServerVFS(t, backing, disableStreaming, tweak, nil) +} + +// newMultipartTestServerVFS is like newMultipartTestServerOpt but also +// overrides the VFS options (nil for the defaults) and disables the named +// features on the backing remote. A backing with features disabled must have +// a unique config string (e.g. a distinct description=) so an active VFS +// wrapping a fully-featured instance of the same remote isn't reused. +func newMultipartTestServerVFS(t *testing.T, backing string, disableStreaming bool, tweak func(*Options), vfsOpt *vfscommon.Options, disableFeatures ...string) (*minio.Core, fs.Fs, string) { fstest.Initialise() ctx := context.Background() + if len(disableFeatures) > 0 { + var ci *fs.ConfigInfo + ctx, ci = fs.AddConfig(ctx) + ci.DisableFeatures = disableFeatures + } if backing == "" { backing = t.TempDir() } @@ -65,10 +79,13 @@ func newMultipartTestServerOpt(t *testing.T, backing string, disableStreaming bo // 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)) + if vfsOpt == nil { + vfsOpt = &vfscommon.Opt + } // 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 { + if root, err := vfs.New(ctx, f, vfsOpt).Root(); err == nil { root.ForgetAll() } @@ -81,7 +98,7 @@ func newMultipartTestServerOpt(t *testing.T, backing string, disableStreaming bo if tweak != nil { tweak(&opt) } - w, err := newServer(ctx, f, &opt, &vfscommon.Opt, &proxy.Opt) + w, err := newServer(ctx, f, &opt, vfsOpt, &proxy.Opt) require.NoError(t, err) go func() { _ = w.Serve() }() t.Cleanup(func() { _ = w.Shutdown() }) @@ -452,32 +469,43 @@ func TestMultipartBufferLimit(t *testing.T) { assert.Equal(t, want, readObject(t, f, bucket, object)) } +// stubSink is a multipartUpload sink which records how it was closed. +type stubSink struct { + closed bool + abortErr error // the reason passed to CloseWithError +} + +func (s *stubSink) Write(p []byte) (int, error) { return len(p), nil } + +func (s *stubSink) Close() error { + s.closed = true + return nil +} + +func (s *stubSink) CloseWithError(err error) error { + s.closed = true + s.abortErr = err + return nil +} + // TestMultipartAbortDuringUploadPart aborts the upload between a part's // waitForTurn and its streamPart - as happens when the abort arrives while // the part body is still being received from the client - and checks that // streamPart fails cleanly instead of panicking on the torn-down upload. func TestMultipartAbortDuringUploadPart(t *testing.T) { up := newMultipartUpload("bucket", "key", "bucket/key", "bucket/key", nil, 0) - - // Mimic CreateMultipartUpload's background PutStream with a stub consumer. - pr, pw := io.Pipe() - _, cancel := context.WithCancel(context.Background()) - defer cancel() - up.pipeW = pw - up.putCancel = cancel - up.putDone = make(chan struct{}) - go func() { - _, err := io.Copy(io.Discard, pr) - up.putErr = err - _ = pr.CloseWithError(err) - close(up.putDone) - }() + sink := &stubSink{} + up.fh = sink // An UploadPart in progress: the part is admitted, then the abort lands // while its body is still being received. contents := []byte("hello") require.NoError(t, up.waitForTurn(1, int64(len(contents)))) - require.NoError(t, up.abort(context.Background())) + require.NoError(t, up.abort()) + + // The abort must abandon the write rather than committing it. + assert.True(t, sink.closed) + assert.Equal(t, errMultipartAborted, sink.abortErr) // The UploadPart resumes: it buffers the part and calls streamPart. rw := multipart.NewRW() @@ -493,6 +521,328 @@ func TestMultipartAbortDuringUploadPart(t *testing.T) { up.mu.Unlock() } +// TestMultipartCloseAfterAbort checks that a close racing an abort reports +// the abort: gofakes3 can dispatch CompleteMultipartUpload and +// AbortMultipartUpload for the same uploadID concurrently, and a Complete +// that loses the race must not report success for data that was never +// committed. +func TestMultipartCloseAfterAbort(t *testing.T) { + up := newMultipartUpload("bucket", "key", "bucket/key", "bucket/key", nil, 0) + up.fh = &stubSink{} + require.NoError(t, up.abort()) + require.ErrorIs(t, up.close(), gofakes3.ErrNoSuchUpload) + + // close stays idempotent after a successful close. + up = newMultipartUpload("bucket", "key", "bucket/key", "bucket/key", nil, 0) + up.fh = &stubSink{} + require.NoError(t, up.close()) + require.NoError(t, up.close()) +} + +// TestMultipartCloseIncomplete checks that close refuses to commit while a +// part is still buffered awaiting an earlier one, leaving the upload open +// for more parts or an abort. +func TestMultipartCloseIncomplete(t *testing.T) { + up := newMultipartUpload("bucket", "key", "bucket/key", "bucket/key", nil, 0) + sink := &stubSink{} + up.fh = sink + + // Part 2 arrives ahead of part 1 so it is buffered, not streamed. + contents := []byte("hello") + rw := multipart.NewRW() + _, err := rw.Write(contents) + require.NoError(t, err) + md5Sum := md5.Sum(contents) + require.NoError(t, up.waitForTurn(2, int64(len(contents)))) + require.NoError(t, up.streamPart(2, int64(len(contents)), md5Sum[:], rw)) + + require.ErrorIs(t, up.close(), gofakes3.ErrInvalidPart) + assert.False(t, sink.closed) + + // The upload is still open so abort tears it down. + require.NoError(t, up.abort()) + assert.True(t, sink.closed) +} + +// failingSink is a sink whose Close fails and which cannot abandon writes, +// like a caching VFS handle whose synchronous write-back fails. +type failingSink struct{} + +func (failingSink) Write(p []byte) (int, error) { return len(p), nil } +func (failingSink) Close() error { return errBoom } + +// TestMultipartAbortAlwaysSucceeds checks that AbortMultipartUpload reports +// success even when closing the sink fails: the upload is torn down either +// way, and an error reply would leave gofakes3's record of the upload alive +// with ours consumed, so every retried abort would 404 on a ghost upload. +func TestMultipartAbortAlwaysSucceeds(t *testing.T) { + b, _, bucket := newPutTestBackend(t, "", nil) + ctx := context.Background() + _vfs, err := b.s.getVFS(ctx) + require.NoError(t, err) + + up := newMultipartUpload(bucket, "key", bucket+"/key", bucket+"/"+multipartUploadPrefix+"x", nil, 0) + up.fh = failingSink{} + up.vfs = _vfs + const uploadID = gofakes3.UploadID("failing-close") + b.multipartUploads.Store(uploadID, up) + + require.NoError(t, b.AbortMultipartUpload(ctx, bucket, "key", uploadID)) + + // The upload record is consumed - a second abort is NoSuchUpload. + require.ErrorIs(t, b.AbortMultipartUpload(ctx, bucket, "key", uploadID), gofakes3.ErrNoSuchUpload) +} + +// TestMultipartCompleteRenameFailureKeepsUpload checks that a Complete +// failing after the commit (here: the rename of the temporary object) keeps +// the upload record, so the retried CompleteMultipartUpload which gofakes3 +// allows on a backend error finds the upload instead of a NoSuchUpload. +func TestMultipartCompleteRenameFailureKeepsUpload(t *testing.T) { + b, _, bucket := newPutTestBackend(t, "", nil) + ctx := context.Background() + _vfs, err := b.s.getVFS(ctx) + require.NoError(t, err) + + // A committed upload whose temporary object is missing: the rename + // fails after the commit succeeded. + up := newMultipartUpload(bucket, "key", bucket+"/key", bucket+"/"+multipartUploadPrefix+"missing", nil, 0) + up.fh = &stubSink{} + up.vfs = _vfs + const uploadID = gofakes3.UploadID("rename-fails") + b.multipartUploads.Store(uploadID, up) + + _, _, err = b.CompleteMultipartUpload(ctx, bucket, "key", uploadID, &gofakes3.CompleteMultipartUploadRequest{}) + require.Error(t, err) + _, err = b.loadUpload(uploadID) + require.NoError(t, err, "the upload record must survive a retryable Complete failure") +} + +// cacheWritesVFSOpt returns VFS options with --vfs-cache-mode writes and the +// given write-back delay. +func cacheWritesVFSOpt(writeBack time.Duration) *vfscommon.Options { + vfsOpt := vfscommon.Opt + vfsOpt.CacheMode = vfscommon.CacheModeWrites + vfsOpt.WriteBack = fs.Duration(writeBack) + return &vfsOpt +} + +// waitForContent waits for bucket/object on the backing Fs to hold want +// (e.g. after the VFS write-back delay). +func waitForContent(t *testing.T, f fs.Fs, bucket, object string, want []byte) { + ctx := context.Background() + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + if o, err := f.NewObject(ctx, path.Join(bucket, object)); err == nil { + if rc, err := o.Open(ctx); err == nil { + got, err := io.ReadAll(rc) + _ = rc.Close() + if err == nil && bytes.Equal(got, want) { + return + } + } + } + time.Sleep(50 * time.Millisecond) + } + t.Fatalf("object %s/%s never reached the expected content on the backing remote", bucket, object) +} + +// TestMultipartCacheModeWrites checks that with --vfs-cache-mode writes a +// multipart upload goes through the VFS cache and is written back to the +// backing remote, with no temporary object left behind. Also run with +// --disable-multipart-streaming, which only affects the streaming path - the +// cache needs no PutStream, so backends without one take this path instead +// of buffering in memory. +func TestMultipartCacheModeWrites(t *testing.T) { + for _, tc := range []struct { + name string + disableStreaming bool + }{ + {"Streaming", false}, + {"NoStreaming", true}, + } { + t.Run(tc.name, func(t *testing.T) { + core, f, bucket := newMultipartTestServerVFS(t, "", tc.disableStreaming, nil, cacheWritesVFSOpt(100*time.Millisecond)) + const object = "cached.bin" + want, err := multipartUploadParts(t, core, bucket, object, []int{120 * 1024, 100 * 1024, 53 * 1024}) + require.NoError(t, err) + waitForContent(t, f, bucket, object, want) + requireOnly(t, f, bucket, object) + }) + } +} + +// TestMultipartCacheModeMinimal checks that --vfs-cache-mode minimal takes +// the cached path just like writes - the parts are written through a +// read-write handle, which the VFS caches from minimal up - including with +// --disable-multipart-streaming set, which only affects the streaming path. +func TestMultipartCacheModeMinimal(t *testing.T) { + vfsOpt := vfscommon.Opt + vfsOpt.CacheMode = vfscommon.CacheModeMinimal + vfsOpt.WriteBack = fs.Duration(100 * time.Millisecond) + core, f, bucket := newMultipartTestServerVFS(t, "", true, nil, &vfsOpt) + const object = "cached-minimal.bin" + + want, err := multipartUploadParts(t, core, bucket, object, []int{120 * 1024, 100 * 1024, 53 * 1024}) + require.NoError(t, err) + waitForContent(t, f, bucket, object, want) + requireOnly(t, f, bucket, object) +} + +// TestMultipartCacheModeWritesAbort checks that with --vfs-cache-mode writes +// an aborted upload is discarded from the cache: nothing reaches the backing +// remote and an existing object at the key survives. +func TestMultipartCacheModeWritesAbort(t *testing.T) { + core, f, bucket := newMultipartTestServerVFS(t, "", false, nil, cacheWritesVFSOpt(100*time.Millisecond)) + ctx := context.Background() + const object = "cached-abort.bin" + + existing := []byte(random.String(100)) + _, err := core.PutObject(ctx, bucket, object, bytes.NewReader(existing), int64(len(existing)), "", "", minio.PutObjectOptions{}) + require.NoError(t, err) + waitForContent(t, f, bucket, object, existing) + + 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)) + + // Wait out several write-back intervals: the aborted upload must not be + // written back, neither over the object nor as a temporary object. + time.Sleep(time.Second) + assert.Equal(t, existing, readObject(t, f, bucket, object)) + requireOnly(t, f, bucket, object) +} + +// TestMultipartCacheModeWritesSupersedesPut checks that a multipart upload +// completed while an earlier PUT to the same key is still in the write-back +// window ends up with the multipart data: both writes go through the same +// cache item, so the earlier PUT's write-back cannot land on top of the +// newer multipart object. +func TestMultipartCacheModeWritesSupersedesPut(t *testing.T) { + core, f, bucket := newMultipartTestServerVFS(t, "", false, nil, cacheWritesVFSOpt(500*time.Millisecond)) + ctx := context.Background() + const object = "supersede.bin" + + // PUT an object; it sits in the cache awaiting write-back. + old := []byte(random.String(100)) + _, err := core.PutObject(ctx, bucket, object, bytes.NewReader(old), int64(len(old)), "", "", minio.PutObjectOptions{}) + require.NoError(t, err) + + // Immediately replace it with a multipart upload to the same key. + want, err := multipartUploadParts(t, core, bucket, object, []int{60 * 1024, 40 * 1024}) + require.NoError(t, err) + + // After all write-backs settle the multipart data must have won. + waitForContent(t, f, bucket, object, want) + time.Sleep(time.Second) + assert.Equal(t, want, readObject(t, f, bucket, object)) + requireOnly(t, f, bucket, object) +} + +// TestMultipartNoPutStream checks that a multipart upload to a remote +// without streaming upload support works with the default cache mode: the +// parts are spooled to a temporary file on local disk and uploaded with a +// known size, rather than being buffered in memory. +func TestMultipartNoPutStream(t *testing.T) { + core, f, bucket := newMultipartTestServerVFS(t, "", false, nil, nil, "PutStream") + require.Nil(t, f.Features().PutStream) + const object = "no-putstream.bin" + + want, err := multipartUploadParts(t, core, bucket, object, []int{120 * 1024, 100 * 1024, 53 * 1024}) + require.NoError(t, err) + assert.Equal(t, want, readObject(t, f, bucket, object)) + requireOnly(t, f, bucket, object) +} + +// TestMultipartNoServerSideMove checks multipart uploads to an atomic remote +// with no server-side move or copy, where the parts stream straight to the +// final object: an upload round-trips, and an aborted upload leaves an +// existing object at the key untouched. +func TestMultipartNoServerSideMove(t *testing.T) { + // The distinct description= gives this remote its own config string so + // it doesn't share a VFS with the fully-featured ":memory:" servers. + core, f, bucket := newMultipartTestServerVFS(t, ":memory,description=no-server-side-move:", false, nil, nil, "Copy") + require.False(t, operations.CanServerSideMove(f)) + ctx := context.Background() + const object = "direct.bin" + + existing := []byte(random.String(100)) + _, err := core.PutObject(ctx, bucket, object, bytes.NewReader(existing), int64(len(existing)), "", "", minio.PutObjectOptions{}) + require.NoError(t, err) + + // An aborted upload must leave the existing object untouched. + 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)) + assert.Equal(t, existing, readObject(t, f, bucket, object)) + requireOnly(t, f, bucket, object) + + // A completed upload replaces it. + 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) +} + +// TestMultipartNoServerSideMovePartialUploads checks that a multipart upload +// to a remote where partial uploads are visible and which has no server-side +// move or copy round-trips: the parts are written straight to the final +// object rather than being buffered in memory. +func TestMultipartNoServerSideMovePartialUploads(t *testing.T) { + core, f, bucket := newMultipartTestServerVFS(t, "", false, nil, nil, "Move", "Copy") + require.False(t, operations.CanServerSideMove(f)) + require.True(t, f.Features().PartialUploads) + const object = "direct-partial.bin" + + 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) +} + +// TestMultipartCacheModeWritesNoServerSideMove checks that with +// --vfs-cache-mode writes a remote with no server-side move or copy is still +// written through the cache rather than falling back to buffering the upload +// in memory, and pins the documented trade-offs of that path: the parts go +// into the cache under the final key, so the in-flight upload is visible +// there, and an aborted upload cannot be abandoned once in the cache, so its +// partial data is written back as if it were a completed object. +func TestMultipartCacheModeWritesNoServerSideMove(t *testing.T) { + // The distinct description= gives this remote its own config string so + // it doesn't share a VFS with the fully-featured ":memory:" servers. + core, f, bucket := newMultipartTestServerVFS(t, ":memory,description=no-move-cache:", false, nil, cacheWritesVFSOpt(100*time.Millisecond), "Copy") + require.False(t, operations.CanServerSideMove(f)) + ctx := context.Background() + + // A round trip goes through the cache under the final key. + const object = "cached-direct.bin" + want, err := multipartUploadParts(t, core, bucket, object, []int{120 * 1024, 100 * 1024, 53 * 1024}) + require.NoError(t, err) + waitForContent(t, f, bucket, object, want) + + // The parts are written to the cache, not buffered in memory, so the + // in-flight upload is visible at the key. + const object2 = "cached-direct-inflight.bin" + uploadID, err := core.NewMultipartUpload(ctx, bucket, object2, minio.PutObjectOptions{}) + require.NoError(t, err) + data := []byte(random.String(50 * 1024)) + _, err = core.PutObjectPart(ctx, bucket, object2, uploadID, 1, bytes.NewReader(data), int64(len(data)), minio.PutObjectPartOptions{}) + require.NoError(t, err) + _, err = core.StatObject(ctx, bucket, object2, minio.StatObjectOptions{}) + assert.NoError(t, err, "in-flight upload should be visible at the key") + + // An aborted upload's partial data is committed to the cache and + // written back. + require.NoError(t, core.AbortMultipartUpload(ctx, bucket, object2, uploadID)) + waitForContent(t, f, bucket, object2, data) + requireOnly(t, f, bucket, object, object2) +} + // TestTempObjectsHiddenFromListings checks that the reserved .rclone_temp_ // prefix, and the multipart prefix used before it was reserved, are hidden // from S3 listings while remaining visible to rclone itself for cleanup. diff --git a/cmd/serve/s3/serve_s3.md b/cmd/serve/s3/serve_s3.md index 2c9152007..23e89ec5d 100644 --- a/cmd/serve/s3/serve_s3.md +++ b/cmd/serve/s3/serve_s3.md @@ -108,22 +108,53 @@ S3 listings but must be removed manually. ### Multipart uploads -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`. - -The upload is atomic so the destination object only ever changes on a +Multipart uploads are written, in part-number order, to a temporary +object which is renamed into place, server-side, on completion, so the +upload is atomic. The object at the key 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. +object already stored under that name and a partly-uploaded object +never becomes visible under it. + +With the default `--vfs-cache-mode off` `serve s3` **streams** each +multipart upload, in part-number order, into a single streaming 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). Remotes that don't support +streaming uploads (those that must know the file size before the +upload starts, such as `onedrive`, `pcloud`, `jottacloud`, `mailru`, +`opendrive`, `putio`, `protondrive` and `zoho`) have the parts spooled +to a temporary file on **local disk** instead, and uploaded with the +size then known on completion, so they need local disk space for the +largest objects in flight rather than memory. + +With `--vfs-cache-mode writes` (or `full`) the parts are written to a +temporary file in the VFS cache and uploaded by the VFS write-back - +see [Multipart uploads and the VFS +cache](#multipart-uploads-and-the-vfs-cache) below. + +The rename into place needs the remote to support a server-side move +or copy, which nearly all do. It is a cheap rename on most remotes, +but on object stores without a real rename (such as `s3` itself) the +move is performed as a server-side copy and delete of the whole +object, which can take time and API calls for large objects. +Concurrent multipart uploads of the same key (which S3 permits) are +safe. Each writes its own temporary object and the last to complete +wins. + +On the few remotes that support neither server side move nor copy, the +parts are written straight to the destination object instead and never +buffered in memory. This is at some cost in atomicity - the incomplete +object is visible under its final name while the upload is in flight, +as it also is for a plain object PUT on such remotes, and concurrent +multipart uploads of the same key write to the same object and can +interleave. A failed or aborted upload still leaves any pre-existing +object untouched provided the remote uploads atomically and the VFS +cache is off; on a remote where partial uploads are visible it may +leave partial data at the key (like a plain PUT there), and with +`--vfs-cache-mode writes` (or `full`) a write to the cache cannot be +abandoned, so an aborted upload's partial data is written back to the +remote as if it had completed. **Features** @@ -138,10 +169,14 @@ also need to support a server-side move or copy. as one continuous stream. - 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). + name untouched, and a partly-uploaded object never becomes visible + (except on the few remotes with no server-side move or copy, as + above). +- Multipart uploads go through the VFS like any other upload, so they + show in rclone's transfer stats and obey `--bwlimit`. +- Backend-agnostic - it only needs the remote to support a server-side + move or copy for the rename into place, which nearly all do; a remote + without streaming upload support spools to local disk as above. **Limitations** @@ -167,14 +202,62 @@ also need to support a server-side move or copy. 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. -- 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_temp_multipart_`) may be left behind; - it is hidden from S3 listings but must be removed manually. + effectively single-threaded. When streaming, the remote's own upload + runs concurrently with the parts arriving; with the local disk spool + or the VFS cache the upload to the remote only starts on completion. +- If `serve s3` is killed part-way through an upload the temporary + object (named with a leading `.rclone_temp_multipart_`) may be left + behind; it is hidden from S3 listings but must be removed manually. + +#### Multipart uploads and the VFS cache + +With `--vfs-cache-mode writes` (or `full`) multipart uploads do not +stream to the remote at all. The parts are written, in part-number +order, to a temporary file in the VFS cache. On completion the file is +renamed into place and uploaded by the VFS write-back, exactly like a +plain object PUT. This needs no streaming upload support from the +remote. The rename normally happens in the cache before the upload has +started, but the VFS requires the remote to support a server-side move +or copy to rename files at all (and uses one if the temporary file has +already been written back, e.g. with `--vfs-write-back 0`). On remotes +without either, the parts are written to the cache directly under the +final key instead: the upload still never touches memory, but it loses +its atomicity - the in-flight upload is visible at the key, and an +aborted upload cannot be abandoned once in the cache, so its partial +data is written back to the remote as if it were a completed object. + +Remotes that benefit from `--vfs-cache-mode writes`: + +- **Remotes over slow or unreliable links.** A failure in a streamed + upload aborts the whole multipart upload and the client must start + again from the first part; a failed write-back upload is retried by + the VFS (see `--vfs-cache-max-age` and friends) without the client + being involved. Ingest from the client also runs at local disk speed + rather than being throttled to the remote's pace. +- **Workloads that read back or overwrite what they just wrote.** The + completed object stays in the cache, so subsequent `GET`/`HEAD` + requests are served locally, and plain PUTs and multipart uploads to + the same key go through the same cache entry so the last write wins + regardless of upload style. + +The trade-offs of the VFS cache: + +- The whole object lands on local disk, so the cache (`--cache-dir`) + needs space for the largest objects in flight; `--vfs-cache-max-size` + cannot evict files which are still being uploaded. +- The `200 OK` for `CompleteMultipartUpload` means the data is safely + in the **local cache**, not yet on the remote - the same durability + the cache gives plain PUTs. If an acknowledgement must mean the data + has reached the remote (for example WAL archiving), use the default + `--vfs-cache-mode off`. +- The upload to the remote only starts on completion, rather than + overlapping with the parts arriving, so the data reaches the remote + later than with streaming. +- If `serve s3` is killed part-way through an upload, the temporary + file survives in the cache and the VFS cache recovery uploads it to + the remote on restart as a temporary object (named with a leading + `.rclone_temp_multipart_`); as with the streaming path, it is + hidden from S3 listings but must be removed manually. #### Cleaning up temporary objects @@ -202,20 +285,18 @@ hidden from listings and can be cleaned up the same way. #### Disabling streaming -If you pass `--disable-multipart-streaming`, or the remote doesn't -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, -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. +If you pass `--disable-multipart-streaming`, 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. 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. This flag is the only thing that makes multipart uploads +buffer in memory - it is never done because of missing remote +capabilities. Consider `--vfs-cache-mode writes` instead, which +buffers the upload in the VFS cache on disk and takes precedence over +`--disable-multipart-streaming`. ### Bugs