f-denkena/rclone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

serve s3: upload all multipart uploads via the VFS

Browse Source 
Multipart uploads used to be streamed directly to the remote with their
own PutStream machinery, bypassing the VFS, a design left over from
before the VFS could abandon a streaming write.

They are now written through the VFS exactly like plain object PUTs in
every cache mode. The parts are written, in part-number order, to a
temporary object which is renamed into place server-side on
completion.

With the default --vfs-cache-mode off the parts stream through the VFS
to the remote as they arrive. With --vfs-cache-mode writes or above
they are buffered in the VFS cache and uploaded by its write-back.

User visible changes:

- Multipart uploads now show in rclone's transfer stats and obey
  --bwlimit (previously they bypassed both).
- Remotes without streaming upload support now spool the upload to a
  temporary file on local disk instead of buffering it in memory.
- Multipart uploads are never buffered in memory because of missing
  remote capabilities - only --disable-multipart-streaming does that.
- Remotes that upload atomically now also write to a temporary object
  renamed into place, so an in-progress multipart upload is no longer
  briefly visible under its final key.
- On the few remotes with no server-side move or copy the parts are
  written straight to the final object in all cache modes.
- With --vfs-cache-mode writes, plain PUTs and multipart uploads to the
  same key go through the same cache entry, so an earlier PUT still in
  the write-back window can no longer be written back over a newer
  multipart upload.
- Failed write-backs are retried by the VFS without the client having
  to restart the upload, and completed objects are served from the
  cache for read-after-write.
This commit is contained in:
Nick Craig-Wood
2026-08-11 20:58:48 +01:00
parent 0aa90200bd
commit b4db289d0a
4 changed files with 655 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/serve/s3/backend.go
+2 -2
View File
@@ -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)
}
}
cmd/serve/s3/multipart.go
+156 -149
View File
@@ -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,44 +27,45 @@ 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
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)
}
@@ -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
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:
cmd/serve/s3/multipart_test.go
+370 -20
View File
@@ -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.
cmd/serve/s3/serve_s3.md
+122 -41
View File
@@ -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