Repository: rclone/rclone
Pull request #4056 (2020-03-11 16:55:13)

cache: move methods used for testing into test file
#### What is the purpose of this change?

Some methods in backend/cache had an all-caps warning that they should only be used in tests. This PR moves those methods to a file that is only compiled in test mode, so the warning is unnecessary.

#### Was the change discussed in an issue or in the forum before?

I don't believe so.

#### Checklist

- [x] I have read the [contribution guidelines](https://github.com/rclone/rclone/blob/master/CONTRIBUTING.md#submitting-a-pull-request).
- [x] I have added tests for all changes in this PR if appropriate.
- [x] I have added documentation for the changes if appropriate.
- [x] All commit messages are in [house style](https://github.com/rclone/rclone/blob/master/CONTRIBUTING.md#commit-messages).
- [x] I'm done, this Pull Request is ready for review :-)



Patch:
diff --git a/backend/b2/upload.go b/backend/b2/upload.go
index 00eeb843d..d390c340b 100644
--- a/backend/b2/upload.go
+++ b/backend/b2/upload.go
@@ -184,13 +184,6 @@ func (up *largeUpload) returnUploadURL(upload *api.GetUploadPartURLResponse) {
 	up.uploadMu.Unlock()
 }
 
-// clearUploadURL clears the current UploadURL and the AuthorizationToken
-func (up *largeUpload) clearUploadURL() {
-	up.uploadMu.Lock()
-	up.uploads = nil
-	up.uploadMu.Unlock()
-}
-
 // Transfer a chunk
 func (up *largeUpload) transferChunk(ctx context.Context, part int64, body []byte) error {
 	err := up.f.pacer.Call(func() (bool, error) {
diff --git a/backend/cache/storage_persistent.go b/backend/cache/storage_persistent.go
index 0c36a4649..054857c24 100644
--- a/backend/cache/storage_persistent.go
+++ b/backend/cache/storage_persistent.go
@@ -980,15 +980,6 @@ func (b *Persistent) updatePendingUpload(remote string, fn func(item *tempUpload
 	})
 }
 
-// SetPendingUploadToStarted is a way to mark an entry as started (even if it's not already)
-// TO BE USED IN TESTING ONLY
-func (b *Persistent) SetPendingUploadToStarted(remote string) error {
-	return b.updatePendingUpload(remote, func(item *tempUploadInfo) error {
-		item.Started = true
-		return nil
-	})
-}
-
 // ReconcileTempUploads will recursively look for all the files in the temp directory and add them to the queue
 func (b *Persistent) ReconcileTempUploads(ctx context.Context, cacheFs *Fs) error {
 	return b.db.Update(func(tx *bolt.Tx) error {
@@ -1036,19 +1027,6 @@ func (b *Persistent) ReconcileTempUploads(ctx context.Context, cacheFs *Fs) erro
 	})
 }
 
-// PurgeTempUploads will remove all the pending uploads from the queue
-// TO BE USED IN TESTING ONLY
-func (b *Persistent) PurgeTempUploads() {
-	b.tempQueueMux.Lock()
-	defer b.tempQueueMux.Unlock()
-
-	_ = b.db.Update(func(tx *bolt.Tx) error {
-		_ = tx.DeleteBucket([]byte(tempBucket))
-		_, _ = tx.CreateBucketIfNotExists([]byte(tempBucket))
-		return nil
-	})
-}
-
 // Close should be called when the program ends gracefully
 func (b *Persistent) Close() {
 	b.cleanupMux.Lock()
diff --git a/backend/drive/drive.go b/backend/drive/drive.go
index feee84a69..928e089aa 100644
--- a/backend/drive/drive.go
+++ b/backend/drive/drive.go
@@ -642,17 +642,21 @@ func containsString(slice []string, s string) bool {
 	return false
 }
 
-// getRootID returns the canonical ID for the "root" ID
-func (f *Fs) getRootID() (string, error) {
-	var info *drive.File
-	var err error
+// getFile returns drive.File for the ID passed and fields passed in
+func (f *Fs) getFile(ID string, fields googleapi.Field) (info *drive.File, err error) {
 	err = f.pacer.CallNoRetry(func() (bool, error) {
-		info, err = f.svc.Files.Get("root").
-			Fields("id").
+		info, err = f.svc.Files.Get(ID).
+			Fields(fields).
 			SupportsAllDrives(true).
 			Do()
 		return f.shouldRetry(err)
 	})
+	return info, err
+}
+
+// getRootID returns the canonical ID for the "root" ID
+func (f *Fs) getRootID() (string, error) {
+	info, err := f.getFile("root", "id")
 	if err != nil {
 		return "", errors.Wrap(err, "couldn't find root directory ID")
 	}
@@ -2042,11 +2046,13 @@ func (f *Fs) Precision() time.Duration {
 func (f *Fs) Copy(ctx context.Context, src fs.Object, remote string) (fs.Object, error) {
 	var srcObj *baseObject
 	ext := ""
+	readDescription := false
 	switch src := src.(type) {
 	case *Object:
 		srcObj = &src.baseObject
 	case *documentObject:
 		srcObj, ext = &src.baseObject, src.ext()
+		readDescription = true
 	case *linkObject:
 		srcObj, ext = &src.baseObject, src.ext()
 	default:
@@ -2070,6 +2076,19 @@ func (f *Fs) Copy(ctx context.Context, src fs.Object, remote string) (fs.Object,
 		return nil, err
 	}
 
+	if readDescription {
+		// preserve the description on copy for docs
+		info, err := f.getFile(srcObj.id, "description")
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to read description for Google Doc")
+		}
+		createInfo.Description = info.Description
+	} else {
+		// don't overwrite the description on copy for files
+		// this should work for docs but it doesn't - it is probably a bug in Google Drive
+		createInfo.Description = ""
+	}
+
 	var info *drive.File
 	err = f.pacer.Call(func() (bool, error) {
 		info, err = f.svc.Files.Copy(srcObj.id, createInfo).
diff --git a/backend/onedrive/onedrive.go b/backend/onedrive/onedrive.go
index 93fa73db1..8ee8f9997 100644
--- a/backend/onedrive/onedrive.go
+++ b/backend/onedrive/onedrive.go
@@ -252,6 +252,16 @@ delete OneNote files or otherwise want them to show up in directory
 listing, set this option.`,
 			Default:  false,
 			Advanced: true,
+		}, {
+			Name:    "server_side_across_configs",
+			Default: false,
+			Help: `Allow server side operations (eg copy) to work across different onedrive configs.
+
+This can be useful if you wish to do a server side copy between two
+different Onedrives.  Note that this isn't enabled by default
+because it isn't easy to tell if it will work between any two
+configurations.`,
+			Advanced: true,
 		}, {
 			Name:     config.ConfigEncoding,
 			Help:     config.ConfigEncodingHelp,
@@ -304,11 +314,12 @@ listing, set this option.`,
 
 // Options defines the configuration for this backend
 type Options struct {
-	ChunkSize          fs.SizeSuffix        `config:"chunk_size"`
-	DriveID            string               `config:"drive_id"`
-	DriveType          string               `config:"drive_type"`
-	ExposeOneNoteFiles bool                 `config:"expose_onenote_files"`
-	Enc                encoder.MultiEncoder `config:"encoding"`
+	ChunkSize               fs.SizeSuffix        `config:"chunk_size"`
+	DriveID                 string               `config:"drive_id"`
+	DriveType               string               `config:"drive_type"`
+	ExposeOneNoteFiles      bool                 `config:"expose_onenote_files"`
+	ServerSideAcrossConfigs bool                 `config:"server_side_across_configs"`
+	Enc                     encoder.MultiEncoder `config:"encoding"`
 }
 
 // Fs represents a remote one drive
@@ -577,6 +588,7 @@ func NewFs(name, root string, m configmap.Mapper) (fs.Fs, error) {
 		CaseInsensitive:         true,
 		ReadMimeType:            true,
 		CanHaveEmptyDirectories: true,
+		ServerSideAcrossConfigs: opt.ServerSideAcrossConfigs,
 	}).Fill(f)
 	f.srv.SetErrorHandler(errorHandler)
 
@@ -989,10 +1001,13 @@ func (f *Fs) Copy(ctx context.Context, src fs.Object, remote string) (fs.Object,
 		return nil, err
 	}
 
-	srcPath := srcObj.rootPath()
-	dstPath := f.rootPath(remote)
-	if strings.ToLower(srcPath) == strings.ToLower(dstPath) {
-		return nil, errors.Errorf("can't copy %q -> %q as are same name when lowercase", srcPath, dstPath)
+	// Check we aren't overwriting a file on the same remote
+	if srcObj.fs == f {
+		srcPath := srcObj.rootPath()
+		dstPath := f.rootPath(remote)
+		if strings.ToLower(srcPath) == strings.ToLower(dstPath) {
+			return nil, errors.Errorf("can't copy %q -> %q as are same name when lowercase", srcPath, dstPath)
+		}
 	}
 
 	// Create temporary object
diff --git a/backend/s3/s3.go b/backend/s3/s3.go
index e62d85ee7..2e1f1d574 100644
--- a/backend/s3/s3.go
+++ b/backend/s3/s3.go
@@ -1099,7 +1099,7 @@ func s3Connection(opt *Options) (*s3.S3, *session.Session, error) {
 		opt.ForcePathStyle = false
 	}
 	awsConfig := aws.NewConfig().
-		WithMaxRetries(fs.Config.LowLevelRetries).
+		WithMaxRetries(0). // Rely on rclone's retry logic
 		WithCredentials(cred).
 		WithHTTPClient(fshttp.NewClient(fs.Config)).
 		WithS3ForcePathStyle(opt.ForcePathStyle).
@@ -1206,17 +1206,12 @@ func NewFs(name, root string, m configmap.Mapper) (fs.Fs, error) {
 		return nil, err
 	}
 
-	pc := fs.NewPacer(pacer.NewS3(pacer.MinSleep(minSleep)))
-	// Set pacer retries to 0 because we are relying on SDK retry mechanism.
-	// Setting it to 1 because in context of pacer it means 1 attempt.
-	pc.SetRetries(1)
-
 	f := &Fs{
 		name:  name,
 		opt:   *opt,
 		c:     c,
 		ses:   ses,
-		pacer: pc,
+		pacer: fs.NewPacer(pacer.NewS3(pacer.MinSleep(minSleep))),
 		cache: bucket.NewCache(),
 		srv:   fshttp.NewClient(fs.Config),
 		pools: make(map[int64]*pool.Pool),
diff --git a/docs/content/authors.md b/docs/content/authors.md
index 2ae2c9688..2fa2a7478 100644
--- a/docs/content/authors.md
+++ b/docs/content/authors.md
@@ -345,3 +345,5 @@ Contributors
   * Robert-André Mauchin <zebob.m@gmail.com>
   * evileye <48332831+ibiruai@users.noreply.github.com>
   * Joachim Brandon LeBlanc <brandon@leblanc.codes>
+  * Patryk Jakuszew <patryk.jakuszew@gmail.com>
+  * fishbullet <shindu666@gmail.com>
diff --git a/docs/content/docs.md b/docs/content/docs.md
index 4a1c341ab..52ba64a1d 100644
--- a/docs/content/docs.md
+++ b/docs/content/docs.md
@@ -879,6 +879,17 @@ This can have a modifier appended with a comma:
 
 - `ascending` or `asc` - order so that the smallest (or oldest) is processed first
 - `descending` or `desc` - order so that the largest (or newest) is processed first
+- `mixed` - order so that the smallest is processed first for some threads and the largest for others
+
+If the modifier is `mixed` then it can have an optional percentage
+(which defaults to `50`), eg `size,mixed,25` which means that 25% of
+the threads should be taking the smallest items and 75% the
+largest. The threads which take the smallest first will always take
+the smallest first and likewise the largest first threads. The `mixed`
+mode can be useful to minimise the transfer time when you are
+transferring a mixture of large and small files - the large files are
+guaranteed upload threads and bandwidth and the small files will be
+processed continuously.
 
 If no modifier is supplied then the order is `ascending`.
 
diff --git a/docs/content/filtering.md b/docs/content/filtering.md
index a593ef908..96bcbfa89 100644
--- a/docs/content/filtering.md
+++ b/docs/content/filtering.md
@@ -17,7 +17,8 @@ Each path as it passes through rclone is matched against the include
 and exclude rules like `--include`, `--exclude`, `--include-from`,
 `--exclude-from`, `--filter`, or `--filter-from`. The simplest way to
 try them out is using the `ls` command, or `--dry-run` together with
-`-v`.
+`-v`. `--filter-from`, `--exclude-from`, `--include-from`, `--files-from`
+understand `-` as a file name to mean read from standard input.
 
 ## Patterns ##
 
diff --git a/docs/content/flags.md b/docs/content/flags.md
index 85a5b460a..ded8129c8 100755
--- a/docs/content/flags.md
+++ b/docs/content/flags.md
@@ -41,13 +41,13 @@ These flags are available for every command.
       --dump-bodies                          Dump HTTP headers and bodies - may contain sensitive info
       --dump-headers                         Dump HTTP headers - may contain sensitive info
       --exclude stringArray                  Exclude files matching pattern
-      --exclude-from stringArray             Read exclude patterns from file
+      --exclude-from stringArray             Read exclude patterns from file (use - to read from stdin).
       --exclude-if-present string            Exclude directories if filename is present
       --expect-continue-timeout duration     Timeout when using expect / 100-continue in HTTP (default 1s)
       --fast-list                            Use recursive list if available. Uses more memory but fewer transactions.
-      --files-from stringArray               Read list of source-file names from file
+      --files-from stringArray               Read list of source-file names from file (use - to read from stdin).
   -f, --filter stringArray                   Add a file-filtering rule
-      --filter-from stringArray              Read filtering patterns from a file
+      --filter-from stringArray              Read filtering patterns from a file (use - to read from stdin).
       --ignore-case                          Ignore case in filters (case insensitive)
       --ignore-case-sync                     Ignore case when synchronizing
       --ignore-checksum                      Skip post copy check of checksums.
@@ -57,7 +57,7 @@ These flags are available for every command.
   -I, --ignore-times                         Don't skip files that match size and time - transfer all files
       --immutable                            Do not modify files. Fail if existing files have been modified.
       --include stringArray                  Include files matching pattern
-      --include-from stringArray             Read include patterns from file
+      --include-from stringArray             Read include patterns from file (use - to read from stdin).
       --log-file string                      Log everything to this file
       --log-format string                    Comma separated list of log format options (default "date,time")
       --log-level string                     Log level DEBUG|INFO|NOTICE|ERROR (default "NOTICE")
diff --git a/fs/accounting/accounting.go b/fs/accounting/accounting.go
index af55e2688..1693ed2e9 100644
--- a/fs/accounting/accounting.go
+++ b/fs/accounting/accounting.go
@@ -171,19 +171,22 @@ func (acc *Account) averageLoop() {
 	}
 }
 
-// Check the read is valid
-func (acc *Account) checkRead() (err error) {
+// Check the read is valid returning the number of bytes it is over
+func (acc *Account) checkRead() (over int64, err error) {
 	acc.statmu.Lock()
-	if acc.max >= 0 && acc.stats.GetBytes() >= acc.max {
-		acc.statmu.Unlock()
-		return ErrorMaxTransferLimitReachedFatal
+	if acc.max >= 0 {
+		over = acc.stats.GetBytes() - acc.max
+		if over >= 0 {
+			acc.statmu.Unlock()
+			return over, ErrorMaxTransferLimitReachedFatal
+		}
 	}
 	// Set start time.
 	if acc.start.IsZero() {
 		acc.start = time.Now()
 	}
 	acc.statmu.Unlock()
-	return nil
+	return over, nil
 }
 
 // ServerSideCopyStart should be called at the start of a server side copy
@@ -223,10 +226,18 @@ func (acc *Account) accountRead(n int) {
 
 // read bytes from the io.Reader passed in and account them
 func (acc *Account) read(in io.Reader, p []byte) (n int, err error) {
-	err = acc.checkRead()
+	_, err = acc.checkRead()
 	if err == nil {
 		n, err = in.Read(p)
 		acc.accountRead(n)
+		if over, checkErr := acc.checkRead(); checkErr == ErrorMaxTransferLimitReachedFatal {
+			// chop the overage off
+			n -= int(over)
+			if n < 0 {
+				n = 0
+			}
+			err = checkErr
+		}
 	}
 	return n, err
 }
@@ -242,7 +253,7 @@ func (acc *Account) Read(p []byte) (n int, err error) {
 func (acc *Account) AccountRead(n int) (err error) {
 	acc.mu.Lock()
 	defer acc.mu.Unlock()
-	err = acc.checkRead()
+	_, err = acc.checkRead()
 	if err == nil {
 		acc.accountRead(n)
 	}
diff --git a/fs/config/config.go b/fs/config/config.go
index c7793f412..53d67ef03 100644
--- a/fs/config/config.go
+++ b/fs/config/config.go
@@ -558,6 +558,7 @@ func saveConfig() error {
 		_ = enc.Close()
 	}
 
+	_ = f.Sync()
 	err = f.Close()
 	if err != nil {
 		return errors.Errorf("Failed to close config file: %v", err)
diff --git a/fs/filter/filter.go b/fs/filter/filter.go
index 460fad6a6..aa52ab010 100644
--- a/fs/filter/filter.go
+++ b/fs/filter/filter.go
@@ -465,12 +465,17 @@ func (f *Filter) IncludeObject(ctx context.Context, o fs.Object) bool {
 //
 // It ignores empty lines and lines starting with '#' or ';'
 func forEachLine(path string, fn func(string) error) (err error) {
-	in, err := os.Open(path)
-	if err != nil {
-		return err
+	var scanner *bufio.Scanner
+	if path == "-" {
+		scanner = bufio.NewScanner(os.Stdin)
+	} else {
+		in, err := os.Open(path)
+		if err != nil {
+			return err
+		}
+		scanner = bufio.NewScanner(in)
+		defer fs.CheckClose(in, &err)
 	}
-	defer fs.CheckClose(in, &err)
-	scanner := bufio.NewScanner(in)
 	for scanner.Scan() {
 		line := scanner.Text()
 		line = strings.TrimSpace(line)
diff --git a/fs/filter/filterflags/filterflags.go b/fs/filter/filterflags/filterflags.go
index 913cf7581..603218b04 100644
--- a/fs/filter/filterflags/filterflags.go
+++ b/fs/filter/filterflags/filterflags.go
@@ -24,13 +24,13 @@ func AddFlags(flagSet *pflag.FlagSet) {
 	rc.AddOptionReload("filter", &Opt, Reload)
 	flags.BoolVarP(flagSet, &Opt.DeleteExcluded, "delete-excluded", "", false, "Delete files on dest excluded from sync")
 	flags.StringArrayVarP(flagSet, &Opt.FilterRule, "filter", "f", nil, "Add a file-filtering rule")
-	flags.StringArrayVarP(flagSet, &Opt.FilterFrom, "filter-from", "", nil, "Read filtering patterns from a file")
+	flags.StringArrayVarP(flagSet, &Opt.FilterFrom, "filter-from", "", nil, "Read filtering patterns from a file (use - to read from stdin)")
 	flags.StringArrayVarP(flagSet, &Opt.ExcludeRule, "exclude", "", nil, "Exclude files matching pattern")
-	flags.StringArrayVarP(flagSet, &Opt.ExcludeFrom, "exclude-from", "", nil, "Read exclude patterns from file")
+	flags.StringArrayVarP(flagSet, &Opt.ExcludeFrom, "exclude-from", "", nil, "Read exclude patterns from file (use - to read from stdin)")
 	flags.StringVarP(flagSet, &Opt.ExcludeFile, "exclude-if-present", "", "", "Exclude directories if filename is present")
 	flags.StringArrayVarP(flagSet, &Opt.IncludeRule, "include", "", nil, "Include files matching pattern")
-	flags.StringArrayVarP(flagSet, &Opt.IncludeFrom, "include-from", "", nil, "Read include patterns from file")
-	flags.StringArrayVarP(flagSet, &Opt.FilesFrom, "files-from", "", nil, "Read list of source-file names from file")
+	flags.StringArrayVarP(flagSet, &Opt.IncludeFrom, "include-from", "", nil, "Read include patterns from file (use - to read from stdin)")
+	flags.StringArrayVarP(flagSet, &Opt.FilesFrom, "files-from", "", nil, "Read list of source-file names from file (use - to read from stdin)")
 	flags.FVarP(flagSet, &Opt.MinAge, "min-age", "", "Only transfer files older than this in s or suffix ms|s|m|h|d|w|M|y")
 	flags.FVarP(flagSet, &Opt.MaxAge, "max-age", "", "Only transfer files younger than this in s or suffix ms|s|m|h|d|w|M|y")
 	flags.FVarP(flagSet, &Opt.MinSize, "min-size", "", "Only transfer files bigger than this in k or suffix b|k|M|G")
diff --git a/fs/log/syslog_unix.go b/fs/log/syslog_unix.go
index b02f35ecf..3c3133e58 100644
--- a/fs/log/syslog_unix.go
+++ b/fs/log/syslog_unix.go
@@ -27,6 +27,14 @@ var (
 		"CRON":     syslog.LOG_CRON,
 		"AUTHPRIV": syslog.LOG_AUTHPRIV,
 		"FTP":      syslog.LOG_FTP,
+		"LOCAL0":   syslog.LOG_LOCAL0,
+		"LOCAL1":   syslog.LOG_LOCAL1,
+		"LOCAL2":   syslog.LOG_LOCAL2,
+		"LOCAL3":   syslog.LOG_LOCAL3,
+		"LOCAL4":   syslog.LOG_LOCAL4,
+		"LOCAL5":   syslog.LOG_LOCAL5,
+		"LOCAL6":   syslog.LOG_LOCAL6,
+		"LOCAL7":   syslog.LOG_LOCAL7,
 	}
 )
 
diff --git a/fs/sync/pipe.go b/fs/sync/pipe.go
index 60da19841..5e2b39b01 100644
--- a/fs/sync/pipe.go
+++ b/fs/sync/pipe.go
@@ -1,11 +1,12 @@
 package sync
 
 import (
-	"container/heap"
 	"context"
+	"strconv"
 	"strings"
 	"sync"
 
+	"github.com/aalpar/deheap"
 	"github.com/pkg/errors"
 	"github.com/rclone/rclone/fs"
 	"github.com/rclone/rclone/fs/fserrors"
@@ -25,20 +26,22 @@ type pipe struct {
 	totalSize int64
 	stats     func(items int, totalSize int64)
 	less      lessFn
+	fraction  int
 }
 
 func newPipe(orderBy string, stats func(items int, totalSize int64), maxBacklog int) (*pipe, error) {
-	less, err := newLess(orderBy)
+	less, fraction, err := newLess(orderBy)
 	if err != nil {
 		return nil, fserrors.FatalError(err)
 	}
 	p := &pipe{
-		c:     make(chan struct{}, maxBacklog),
-		stats: stats,
-		less:  less,
+		c:        make(chan struct{}, maxBacklog),
+		stats:    stats,
+		less:     less,
+		fraction: fraction,
 	}
 	if p.less != nil {
-		heap.Init(p)
+		deheap.Init(p)
 	}
 	return p, nil
 }
@@ -73,9 +76,6 @@ func (p *pipe) Pop() interface{} {
 	return item
 }
 
-// Check interface satisfied
-var _ heap.Interface = (*pipe)(nil)
-
 // Put an pair into the pipe
 //
 // It returns ok = false if the context was cancelled
@@ -90,7 +90,7 @@ func (p *pipe) Put(ctx context.Context, pair fs.ObjectPair) (ok bool) {
 		// no order-by
 		p.queue = append(p.queue, pair)
 	} else {
-		heap.Push(p, pair)
+		deheap.Push(p, pair)
 	}
 	size := pair.Src.Size()
 	if size > 0 {
@@ -108,9 +108,12 @@ func (p *pipe) Put(ctx context.Context, pair fs.ObjectPair) (ok bool) {
 
 // Get a pair from the pipe
 //
+// If fraction is > the mixed fraction set in the pipe then it gets it
+// from the other end of the heap if order-by is in effect
+//
 // It returns ok = false if the context was cancelled or Close() has
 // been called.
-func (p *pipe) Get(ctx context.Context) (pair fs.ObjectPair, ok bool) {
+func (p *pipe) GetMax(ctx context.Context, fraction int) (pair fs.ObjectPair, ok bool) {
 	if ctx.Err() != nil {
 		return
 	}
@@ -128,8 +131,10 @@ func (p *pipe) Get(ctx context.Context) (pair fs.ObjectPair, ok bool) {
 		pair = p.queue[0]
 		p.queue[0] = fs.ObjectPair{} // avoid memory leak
 		p.queue = p.queue[1:]
+	} else if p.fraction < 0 || fraction < p.fraction {
+		pair = deheap.Pop(p).(fs.ObjectPair)
 	} else {
-		pair = heap.Pop(p).(fs.ObjectPair)
+		pair = deheap.PopMax(p).(fs.ObjectPair)
 	}
 	size := pair.Src.Size()
 	if size > 0 {
@@ -143,6 +148,14 @@ func (p *pipe) Get(ctx context.Context) (pair fs.ObjectPair, ok bool) {
 	return pair, true
 }
 
+// Get a pair from the pipe
+//
+// It returns ok = false if the context was cancelled or Close() has
+// been called.
+func (p *pipe) Get(ctx context.Context) (pair fs.ObjectPair, ok bool) {
+	return p.GetMax(ctx, -1)
+}
+
 // Stats reads the number of items in the queue and the totalSize
 func (p *pipe) Stats() (items int, totalSize int64) {
 	p.mu.Lock()
@@ -163,14 +176,12 @@ func (p *pipe) Close() {
 
 // newLess returns a less function for the heap comparison or nil if
 // one is not required
-func newLess(orderBy string) (less lessFn, err error) {
+func newLess(orderBy string) (less lessFn, fraction int, err error) {
+	fraction = -1
 	if orderBy == "" {
-		return nil, nil
+		return nil, fraction, nil
 	}
 	parts := strings.Split(strings.ToLower(orderBy), ",")
-	if len(parts) > 2 {
-		return nil, errors.Errorf("bad --order-by string %q", orderBy)
-	}
 	switch parts[0] {
 	case "name":
 		less = func(a, b fs.ObjectPair) bool {
@@ -186,7 +197,7 @@ func newLess(orderBy string) (less lessFn, err error) {
 			return a.Src.ModTime(ctx).Before(b.Src.ModTime(ctx))
 		}
 	default:
-		return nil, errors.Errorf("unknown --order-by comparison %q", parts[0])
+		return nil, fraction, errors.Errorf("unknown --order-by comparison %q", parts[0])
 	}
 	descending := false
 	if len(parts) > 1 {
@@ -194,15 +205,27 @@ func newLess(orderBy string) (less lessFn, err error) {
 		case "ascending", "asc":
 		case "descending", "desc":
 			descending = true
+		case "mixed":
+			fraction = 50
+			if len(parts) > 2 {
+				fraction, err = strconv.Atoi(parts[2])
+				if err != nil {
+					return nil, fraction, errors.Errorf("bad mixed fraction --order-by %q", parts[2])
+				}
+			}
+
 		default:
-			return nil, errors.Errorf("unknown --order-by sort direction %q", parts[1])
+			return nil, fraction, errors.Errorf("unknown --order-by sort direction %q", parts[1])
 		}
 	}
+	if (fraction >= 0 && len(parts) > 3) || (fraction < 0 && len(parts) > 2) {
+		return nil, fraction, errors.Errorf("bad --order-by string %q", orderBy)
+	}
 	if descending {
 		oldLess := less
 		less = func(a, b fs.ObjectPair) bool {
 			return !oldLess(a, b)
 		}
 	}
-	return less, nil
+	return less, fraction, nil
 }
diff --git a/fs/sync/sync.go b/fs/sync/sync.go
index c3887c9e6..8bab3bb97 100644
--- a/fs/sync/sync.go
+++ b/fs/sync/sync.go
@@ -241,10 +241,10 @@ func (s *syncCopyMove) currentError() error {
 // pairChecker reads Objects~s on in send to out if they need transferring.
 //
 // FIXME potentially doing lots of hashes at once
-func (s *syncCopyMove) pairChecker(in *pipe, out *pipe, wg *sync.WaitGroup) {
+func (s *syncCopyMove) pairChecker(in *pipe, out *pipe, fraction int, wg *sync.WaitGroup) {
 	defer wg.Done()
 	for {
-		pair, ok := in.Get(s.ctx)
+		pair, ok := in.GetMax(s.ctx, fraction)
 		if !ok {
 			return
 		}
@@ -297,10 +297,10 @@ func (s *syncCopyMove) pairChecker(in *pipe, out *pipe, wg *sync.WaitGroup) {
 
 // pairRenamer reads Objects~s on in and attempts to rename them,
 // otherwise it sends them out if they need transferring.
-func (s *syncCopyMove) pairRenamer(in *pipe, out *pipe, wg *sync.WaitGroup) {
+func (s *syncCopyMove) pairRenamer(in *pipe, out *pipe, fraction int, wg *sync.WaitGroup) {
 	defer wg.Done()
 	for {
-		pair, ok := in.Get(s.ctx)
+		pair, ok := in.GetMax(s.ctx, fraction)
 		if !ok {
 			return
 		}
@@ -316,11 +316,11 @@ func (s *syncCopyMove) pairRenamer(in *pipe, out *pipe, wg *sync.WaitGroup) {
 }
 
 // pairCopyOrMove reads Objects on in and moves or copies them.
-func (s *syncCopyMove) pairCopyOrMove(ctx context.Context, in *pipe, fdst fs.Fs, wg *sync.WaitGroup) {
+func (s *syncCopyMove) pairCopyOrMove(ctx context.Context, in *pipe, fdst fs.Fs, fraction int, wg *sync.WaitGroup) {
 	defer wg.Done()
 	var err error
 	for {
-		pair, ok := in.Get(s.ctx)
+		pair, ok := in.GetMax(s.ctx, fraction)
 		if !ok {
 			return
 		}
@@ -338,7 +338,8 @@ func (s *syncCopyMove) pairCopyOrMove(ctx context.Context, in *pipe, fdst fs.Fs,
 func (s *syncCopyMove) startCheckers() {
 	s.checkerWg.Add(fs.Config.Checkers)
 	for i := 0; i < fs.Config.Checkers; i++ {
-		go s.pairChecker(s.toBeChecked, s.toBeUploaded, &s.checkerWg)
+		fraction := (100 * i) / fs.Config.Checkers
+		go s.pairChecker(s.toBeChecked, s.toBeUploaded, fraction, &s.checkerWg)
 	}
 }
 
@@ -353,7 +354,8 @@ func (s *syncCopyMove) stopCheckers() {
 func (s *syncCopyMove) startTransfers() {
 	s.transfersWg.Add(fs.Config.Transfers)
 	for i := 0; i < fs.Config.Transfers; i++ {
-		go s.pairCopyOrMove(s.ctx, s.toBeUploaded, s.fdst, &s.transfersWg)
+		fraction := (100 * i) / fs.Config.Transfers
+		go s.pairCopyOrMove(s.ctx, s.toBeUploaded, s.fdst, fraction, &s.transfersWg)
 	}
 }
 
@@ -371,7 +373,8 @@ func (s *syncCopyMove) startRenamers() {
 	}
 	s.renamerWg.Add(fs.Config.Checkers)
 	for i := 0; i < fs.Config.Checkers; i++ {
-		go s.pairRenamer(s.toBeRenamed, s.toBeUploaded, &s.renamerWg)
+		fraction := (100 * i) / fs.Config.Checkers
+		go s.pairRenamer(s.toBeRenamed, s.toBeUploaded, fraction, &s.renamerWg)
 	}
 }
 
diff --git a/go.mod b/go.mod
index 5104f598e..c8cf6fb93 100644
--- a/go.mod
+++ b/go.mod
@@ -8,6 +8,7 @@ require (
 	github.com/Azure/go-autorest/autorest/adal v0.6.0 // indirect
 	github.com/Unknwon/goconfig v0.0.0-20191126170842-860a72fb44fd
 	github.com/a8m/tree v0.0.0-20181222104329-6a0b80129de4
+	github.com/aalpar/deheap v0.0.0-20191229192855-f837f7a9ba26
 	github.com/abbot/go-http-auth v0.4.0
 	github.com/anacrolix/dms v1.1.0
 	github.com/atotto/clipboard v0.1.2
diff --git a/go.sum b/go.sum
index fba59e59f..bde8a5e86 100644
--- a/go.sum
+++ b/go.sum
@@ -48,6 +48,8 @@ github.com/Unknwon/goconfig v0.0.0-20191126170842-860a72fb44fd h1:+CYOsXi89xOqBk
 github.com/Unknwon/goconfig v0.0.0-20191126170842-860a72fb44fd/go.mod h1:wngxua9XCNjvHjDiTiV26DaKDT+0c63QR6H5hjVUUxw=
 github.com/a8m/tree v0.0.0-20181222104329-6a0b80129de4 h1:mK1/QgFPU4osbhjJ26B1w738kjQHaGJcon8uCLMS8fk=
 github.com/a8m/tree v0.0.0-20181222104329-6a0b80129de4/go.mod h1:FSdwKX97koS5efgm8WevNf7XS3PqtyFkKDDXrz778cg=
+github.com/aalpar/deheap v0.0.0-20191229192855-f837f7a9ba26 h1:vn0bf9tSbwrl9aBeG5hqGPpWeZGEybrB3LUJkHP/x6w=
+github.com/aalpar/deheap v0.0.0-20191229192855-f837f7a9ba26/go.mod h1:EJFoWbcEEVK22GYKONJjtMNamGYe6p+3x1Pr6zV5gFs=
 github.com/abbot/go-http-auth v0.4.0 h1:QjmvZ5gSC7jm3Zg54DqWE/T5m1t2AfDu6QlXJT0EVT0=
 github.com/abbot/go-http-auth v0.4.0/go.mod h1:Cz6ARTIzApMJDzh5bRMSUou6UMSp0IEXg9km/ci7TJM=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
@@ -576,6 +578,7 @@ gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLks
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
diff --git a/vendor/github.com/aalpar/deheap/LICENSE.txt b/vendor/github.com/aalpar/deheap/LICENSE.txt
new file mode 100644
index 000000000..d8ef07fed
--- /dev/null
+++ b/vendor/github.com/aalpar/deheap/LICENSE.txt
@@ -0,0 +1,20 @@
+Copyright 2019 Aaron H. Alpar
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files
+(the "Software"), to deal in the Software without restriction,
+including without limitation the rights to use, copy, modify, merge,
+publish, distribute, sublicense, and/or sell copies of the Software,
+and to permit persons to whom the Software is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included
+in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
diff --git a/vendor/github.com/aalpar/deheap/README.md b/vendor/github.com/aalpar/deheap/README.md
new file mode 100644
index 000000000..a3a24f0ca
--- /dev/null
+++ b/vendor/github.com/aalpar/deheap/README.md
@@ -0,0 +1,14 @@
+# deheap
+
+Package deheap provides the implementation of a doubly ended heap.
+Doubly ended heaps are heaps with two sides, a min side and a max side.
+Like normal single-sided heaps, elements can be pushed onto and pulled
+off of a deheap.  deheaps have an additional `Pop` function, `PopMax`, that
+returns elements from the opposite side of the ordering.
+
+This implementation has emphasized compatibility with existing libraries
+in the sort and heap packages.
+
+Performace of the deheap functions should be very close to the
+performance of the functions of the heap library
+
diff --git a/vendor/github.com/aalpar/deheap/deheap.go b/vendor/github.com/aalpar/deheap/deheap.go
new file mode 100644
index 000000000..6f53f6fea
--- /dev/null
+++ b/vendor/github.com/aalpar/deheap/deheap.go
@@ -0,0 +1,228 @@
+//
+// Copyright 2019 Aaron H. Alpar
+//
+// Permission is hereby granted, free of charge, to any person obtaining
+// a copy of this software and associated documentation files
+// (the "Software"), to deal in the Software without restriction,
+// including without limitation the rights to use, copy, modify, merge,
+// publish, distribute, sublicense, and/or sell copies of the Software,
+// and to permit persons to whom the Software is furnished to do so,
+// subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included
+// in all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+// MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+// IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+// TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+// SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+//
+
+//
+// Package deheap provides the implementation of a doubly ended heap.
+// Doubly ended heaps are heaps with two sides, a min side and a max side.
+// Like normal single-sided heaps, elements can be pushed onto and pulled
+// off of a deheap.  deheaps have an additional Pop function, PopMax, that
+// returns elements from the opposite side of the ordering.
+//
+// This implementation has emphasized compatibility with existing libraries
+// in the sort and heap packages.
+//
+// Performace of the deheap functions should be very close to the
+// performance of the functions of the heap library
+//
+package deheap
+
+import (
+	"container/heap"
+	"math/bits"
+)
+
+func hparent(i int) int {
+	return (i - 1) / 2
+}
+
+func hlchild(i int) int {
+	return (i * 2) + 1
+}
+
+func parent(i int) int {
+	return ((i + 1) / 4) - 1
+}
+
+func lchild(i int) int {
+	return ((i + 1 ) * 4) - 1
+}
+
+func level(i int) int {
+	return bits.Len(uint(i)+1) - 1
+}
+
+func isMinHeap(i int) bool {
+	return level(i) % 2 == 0
+}
+
+func min4(h heap.Interface, l int, min bool, i int)  int {
+	q := i
+	i++
+	if i >= l {
+		return q
+	}
+	if min == h.Less(i, q) {
+		q = i
+	}
+	i++
+	if i >= l {
+		return q
+	}
+	if min == h.Less(i, q) {
+		q = i
+	}
+	i++
+	if i >= l {
+		return q
+	}
+	if min == h.Less(i, q) {
+		q = i
+	}
+	return q
+}
+
+// min2
+func min2(h heap.Interface, l int, min bool, i int) int {
+	if i+1 >= l {
+		return i
+	}
+	if min != h.Less(i+1, i) {
+		return i
+	}
+	return i + 1
+}
+
+// min3
+func min3(h heap.Interface, l int, min bool, i, j, k int) int {
+	q := i
+	if j < l && h.Less(j, q) == min {
+		q = j
+	}
+	if k < l && h.Less(k, q) == min {
+		q = k
+	}
+	return q
+}
+
+// bubbledown
+func bubbledown(h heap.Interface, l int, min bool, i int) (q int, r int) {
+	q = i
+	r = i
+	for {
+		// find min of children
+		j := min2(h, l, min, hlchild(i))
+		if j >= l {
+			break
+		}
+		// find min of grandchildren
+		k := min4(h, l, min, lchild(i))
+		// swap of less than the element at i
+		v := min3(h, l, min, i, j, k)
+		if v == i || v >= l {
+			break
+		}
+		// v == k
+		q = v
+		h.Swap(v, i)
+		if v == j {
+			break
+		}
+		p := hparent(v)
+		if h.Less(p, v) == min {
+			h.Swap(p, v)
+			r = p
+		}
+		i = v
+	}
+	return q, r
+}
+
+// bubbleup
+func bubbleup(h heap.Interface, min bool, i int) (q bool) {
+	if i < 0 {
+		return false
+	}
+	j := parent(i)
+	for j >= 0 && min == h.Less(i, j) {
+		q = true
+		h.Swap(i, j)
+		i = j
+		j = parent(i)
+	}
+	min = !min
+	j = hparent(i)
+	for j >= 0 && min == h.Less(i, j) {
+		q = true
+		h.Swap(i, j)
+		i = j
+		j = parent(i)
+	}
+	return q
+}
+
+// Pop the smallest value off the heap.  See heap.Pop().
+// Time complexity is O(log n), where n = h.Len()
+func Pop(h heap.Interface) interface{} {
+	l := h.Len()-1
+	h.Swap(0, l)
+	q := h.Pop()
+	bubbledown(h, l, true, 0)
+	return q
+}
+
+// Pop the largest value off the heap.  See heap.Pop().
+// Time complexity is O(log n), where n = h.Len()
+func PopMax(h heap.Interface) interface{} {
+	l := h.Len()
+	j := 0
+	if l > 1 {
+		j = min2(h, l,false, 1)
+	}
+	l = l - 1
+	h.Swap(j, l)
+	q := h.Pop()
+	bubbledown(h, l,false, j)
+	return q
+}
+
+// Remove element at index i.  See heap.Remove().
+// The complexity is O(log n) where n = h.Len().
+func Remove(h heap.Interface, i int) (q interface{}) {
+	l := h.Len() - 1
+	h.Swap(i, l)
+	q = h.Pop()
+	if l != i {
+		q, r := bubbledown(h, l, isMinHeap(i), i)
+		bubbleup(h, isMinHeap(q), q)
+		bubbleup(h, isMinHeap(r), r)
+	}
+	return q
+}
+
+// Push an element onto the heap.  See heap.Push()
+// Time complexity is O(log n), where n = h.Len()
+func Push(h heap.Interface, o interface{}) {
+	h.Push(o)
+	l := h.Len()
+	i := l - 1
+	bubbleup(h, isMinHeap(i), i)
+}
+
+// Init initializes the heap.
+// This should be called once on non-empty heaps before calling Pop(), PopMax() or Push().  See heap.Init()
+func Init(h heap.Interface) {
+	l := h.Len()
+	for i := 0; i < l; i++ {
+		bubbleup(h, isMinHeap(i), i)
+	}
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index a7e8e167b..01d66f5d5 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -12,6 +12,8 @@ github.com/Azure/azure-storage-blob-go/azblob
 github.com/Unknwon/goconfig
 # github.com/a8m/tree v0.0.0-20181222104329-6a0b80129de4
 github.com/a8m/tree
+# github.com/aalpar/deheap v0.0.0-20191229192855-f837f7a9ba26
+github.com/aalpar/deheap
 # github.com/abbot/go-http-auth v0.4.0
 github.com/abbot/go-http-auth
 # github.com/anacrolix/dms v1.1.0


Test patch:
diff --git a/backend/cache/utils_test.go b/backend/cache/utils_test.go
new file mode 100644
index 000000000..c02d86ce9
--- /dev/null
+++ b/backend/cache/utils_test.go
@@ -0,0 +1,23 @@
+package cache
+
+import bolt "go.etcd.io/bbolt"
+
+// PurgeTempUploads will remove all the pending uploads from the queue
+func (b *Persistent) PurgeTempUploads() {
+	b.tempQueueMux.Lock()
+	defer b.tempQueueMux.Unlock()
+
+	_ = b.db.Update(func(tx *bolt.Tx) error {
+		_ = tx.DeleteBucket([]byte(tempBucket))
+		_, _ = tx.CreateBucketIfNotExists([]byte(tempBucket))
+		return nil
+	})
+}
+
+// SetPendingUploadToStarted is a way to mark an entry as started (even if it's not already)
+func (b *Persistent) SetPendingUploadToStarted(remote string) error {
+	return b.updatePendingUpload(remote, func(item *tempUploadInfo) error {
+		item.Started = true
+		return nil
+	})
+}
diff --git a/fs/accounting/accounting_test.go b/fs/accounting/accounting_test.go
index 083072e59..4b876e1cd 100644
--- a/fs/accounting/accounting_test.go
+++ b/fs/accounting/accounting_test.go
@@ -215,8 +215,8 @@ func TestAccountMaxTransfer(t *testing.T) {
 	assert.Equal(t, 10, n)
 	assert.NoError(t, err)
 	n, err = acc.Read(b)
-	assert.Equal(t, 10, n)
-	assert.NoError(t, err)
+	assert.Equal(t, 5, n)
+	assert.Equal(t, ErrorMaxTransferLimitReachedFatal, err)
 	n, err = acc.Read(b)
 	assert.Equal(t, 0, n)
 	assert.Equal(t, ErrorMaxTransferLimitReachedFatal, err)
diff --git a/fs/filter/filter_test.go b/fs/filter/filter_test.go
index 773b3eb04..0d1b4e90f 100644
--- a/fs/filter/filter_test.go
+++ b/fs/filter/filter_test.go
@@ -517,7 +517,7 @@ func TestFilterAddDirRuleOrFileRule(t *testing.T) {
 	}
 }
 
-func TestFilterForEachLine(t *testing.T) {
+func testFilterForEachLine(t *testing.T, useStdin bool) {
 	file := testFile(t, `; comment
 one
 # another comment
@@ -534,7 +534,19 @@ five
 		require.NoError(t, err)
 	}()
 	lines := []string{}
-	err := forEachLine(file, func(s string) error {
+	fileName := file
+	if useStdin {
+		in, err := os.Open(file)
+		require.NoError(t, err)
+		oldStdin := os.Stdin
+		os.Stdin = in
+		defer func() {
+			os.Stdin = oldStdin
+			_ = in.Close()
+		}()
+		fileName = "-"
+	}
+	err := forEachLine(fileName, func(s string) error {
 		lines = append(lines, s)
 		return nil
 	})
@@ -542,6 +554,14 @@ five
 	assert.Equal(t, "one,two,three,four,five,six", strings.Join(lines, ","))
 }
 
+func TestFilterForEachLine(t *testing.T) {
+	testFilterForEachLine(t, false)
+}
+
+func TestFilterForEachLineStdin(t *testing.T) {
+	testFilterForEachLine(t, true)
+}
+
 func TestFilterMatchesFromDocs(t *testing.T) {
 	for _, test := range []struct {
 		glob       string
diff --git a/fs/operations/operations_test.go b/fs/operations/operations_test.go
index 192dd6610..20d00e4af 100644
--- a/fs/operations/operations_test.go
+++ b/fs/operations/operations_test.go
@@ -44,6 +44,7 @@ import (
 	"github.com/rclone/rclone/fs/hash"
 	"github.com/rclone/rclone/fs/operations"
 	"github.com/rclone/rclone/fstest"
+	"github.com/rclone/rclone/lib/random"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -1541,45 +1542,59 @@ func TestCopyFileMaxTransfer(t *testing.T) {
 		accounting.Stats(context.Background()).ResetCounters()
 	}()
 
+	ctx := context.Background()
+
+	const sizeCutoff = 2048
 	file1 := r.WriteFile("file1", "file1 contents", t1)
-	file2 := r.WriteFile("file2", "file2 contents...........", t2)
+	file2 := r.WriteFile("file2", "file2 contents"+random.String(sizeCutoff), t2)
 
 	rfile1 := file1
 	rfile1.Path = "sub/file1"
-	rfile2 := file2
-	rfile2.Path = "sub/file2"
-
-	fs.Config.MaxTransfer = 15
+	rfile2a := file2
+	rfile2a.Path = "sub/file2a"
+	rfile2b := file2
+	rfile2b.Path = "sub/file2b"
+	rfile2c := file2
+	rfile2c.Path = "sub/file2c"
+
+	fs.Config.MaxTransfer = sizeCutoff
 	fs.Config.CutoffMode = fs.CutoffModeHard
-	accounting.Stats(context.Background()).ResetCounters()
+	accounting.Stats(ctx).ResetCounters()
 
-	err := operations.CopyFile(context.Background(), r.Fremote, r.Flocal, rfile1.Path, file1.Path)
+	err := operations.CopyFile(ctx, r.Fremote, r.Flocal, rfile1.Path, file1.Path)
 	require.NoError(t, err)
 	fstest.CheckItems(t, r.Flocal, file1, file2)
 	fstest.CheckItems(t, r.Fremote, rfile1)
 
-	accounting.Stats(context.Background()).ResetCounters()
+	accounting.Stats(ctx).ResetCounters()
 
-	err = operations.CopyFile(context.Background(), r.Fremote, r.Flocal, rfile2.Path, file2.Path)
-	fstest.CheckItems(t, r.Flocal, file1, file2)
-	fstest.CheckItems(t, r.Fremote, rfile1)
+	err = operations.CopyFile(ctx, r.Fremote, r.Flocal, rfile2a.Path, file2.Path)
+	require.NotNil(t, err)
 	assert.Contains(t, err.Error(), "Max transfer limit reached")
 	assert.True(t, fserrors.IsFatalError(err))
+	fstest.CheckItems(t, r.Flocal, file1, file2)
+	fstest.CheckItems(t, r.Fremote, rfile1)
 
 	fs.Config.CutoffMode = fs.CutoffModeCautious
-	accounting.Stats(context.Background()).ResetCounters()
+	accounting.Stats(ctx).ResetCounters()
 
-	err = operations.CopyFile(context.Background(), r.Fremote, r.Flocal, rfile2.Path, file2.Path)
-	fstest.CheckItems(t, r.Flocal, file1, file2)
-	fstest.CheckItems(t, r.Fremote, rfile1)
+	err = operations.CopyFile(ctx, r.Fremote, r.Flocal, rfile2b.Path, file2.Path)
+	require.NotNil(t, err)
 	assert.Contains(t, err.Error(), "Max transfer limit reached")
 	assert.True(t, fserrors.IsFatalError(err))
+	fstest.CheckItems(t, r.Flocal, file1, file2)
+	fstest.CheckItems(t, r.Fremote, rfile1)
+
+	if strings.HasPrefix(r.Fremote.Name(), "TestChunker") {
+		t.Log("skipping remainder of test for chunker as it involves multiple transfers")
+		return
+	}
 
 	fs.Config.CutoffMode = fs.CutoffModeSoft
-	accounting.Stats(context.Background()).ResetCounters()
+	accounting.Stats(ctx).ResetCounters()
 
-	err = operations.CopyFile(context.Background(), r.Fremote, r.Flocal, rfile2.Path, file2.Path)
+	err = operations.CopyFile(ctx, r.Fremote, r.Flocal, rfile2c.Path, file2.Path)
 	require.NoError(t, err)
 	fstest.CheckItems(t, r.Flocal, file1, file2)
-	fstest.CheckItems(t, r.Fremote, rfile1, rfile2)
+	fstest.CheckItems(t, r.Fremote, rfile1, rfile2c)
 }
diff --git a/fs/sync/pipe_test.go b/fs/sync/pipe_test.go
index 1ef34f27d..84c9129e5 100644
--- a/fs/sync/pipe_test.go
+++ b/fs/sync/pipe_test.go
@@ -1,6 +1,7 @@
 package sync
 
 import (
+	"container/heap"
 	"context"
 	"sync"
 	"sync/atomic"
@@ -12,6 +13,9 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// Check interface satisfied
+var _ heap.Interface = (*pipe)(nil)
+
 func TestPipe(t *testing.T) {
 	var queueLength int
 	var queueSize int64
@@ -139,32 +143,37 @@ func TestPipeOrderBy(t *testing.T) {
 		orderBy  string
 		swapped1 bool
 		swapped2 bool
+		fraction int
 	}{
-		{"", false, true},
-		{"size", false, false},
-		{"name", true, true},
-		{"modtime", false, true},
-		{"size,ascending", false, false},
-		{"name,asc", true, true},
-		{"modtime,ascending", false, true},
-		{"size,descending", true, true},
-		{"name,desc", false, false},
-		{"modtime,descending", true, false},
+		{"", false, true, -1},
+		{"size", false, false, -1},
+		{"name", true, true, -1},
+		{"modtime", false, true, -1},
+		{"size,ascending", false, false, -1},
+		{"name,asc", true, true, -1},
+		{"modtime,ascending", false, true, -1},
+		{"size,descending", true, true, -1},
+		{"name,desc", false, false, -1},
+		{"modtime,descending", true, false, -1},
+		{"size,mixed,50", false, false, 25},
+		{"size,mixed,51", true, true, 75},
 	} {
 		t.Run(test.orderBy, func(t *testing.T) {
 			p, err := newPipe(test.orderBy, stats, 10)
 			require.NoError(t, err)
 
-			ok := p.Put(ctx, pair1)
-			assert.True(t, ok)
-			ok = p.Put(ctx, pair2)
-			assert.True(t, ok)
-
 			readAndCheck := func(swapped bool) {
-				readFirst, ok := p.Get(ctx)
-				assert.True(t, ok)
-				readSecond, ok := p.Get(ctx)
-				assert.True(t, ok)
+				var readFirst, readSecond fs.ObjectPair
+				var ok1, ok2 bool
+				if test.fraction < 0 {
+					readFirst, ok1 = p.Get(ctx)
+					readSecond, ok2 = p.Get(ctx)
+				} else {
+					readFirst, ok1 = p.GetMax(ctx, test.fraction)
+					readSecond, ok2 = p.GetMax(ctx, test.fraction)
+				}
+				assert.True(t, ok1)
+				assert.True(t, ok2)
 
 				if swapped {
 					assert.True(t, readFirst == pair2 && readSecond == pair1)
@@ -173,6 +182,11 @@ func TestPipeOrderBy(t *testing.T) {
 				}
 			}
 
+			ok := p.Put(ctx, pair1)
+			assert.True(t, ok)
+			ok = p.Put(ctx, pair2)
+			assert.True(t, ok)
+
 			readAndCheck(test.swapped1)
 
 			// insert other way round
@@ -189,25 +203,37 @@ func TestPipeOrderBy(t *testing.T) {
 
 func TestNewLess(t *testing.T) {
 	t.Run("blankOK", func(t *testing.T) {
-		less, err := newLess("")
+		less, _, err := newLess("")
 		require.NoError(t, err)
 		assert.Nil(t, less)
 	})
 
 	t.Run("tooManyParts", func(t *testing.T) {
-		_, err := newLess("too,many,parts")
+		_, _, err := newLess("size,asc,toomanyparts")
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "bad --order-by string")
+	})
+
+	t.Run("tooManyParts2", func(t *testing.T) {
+		_, _, err := newLess("size,mixed,50,toomanyparts")
 		require.Error(t, err)
 		assert.Contains(t, err.Error(), "bad --order-by string")
 	})
 
+	t.Run("badMixed", func(t *testing.T) {
+		_, _, err := newLess("size,mixed,32.7")
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "bad mixed fraction")
+	})
+
 	t.Run("unknownComparison", func(t *testing.T) {
-		_, err := newLess("potato")
+		_, _, err := newLess("potato")
 		require.Error(t, err)
 		assert.Contains(t, err.Error(), "unknown --order-by comparison")
 	})
 
 	t.Run("unknownSortDirection", func(t *testing.T) {
-		_, err := newLess("name,sideways")
+		_, _, err := newLess("name,sideways")
 		require.Error(t, err)
 		assert.Contains(t, err.Error(), "unknown --order-by sort direction")
 	})
@@ -223,19 +249,23 @@ func TestNewLess(t *testing.T) {
 		orderBy        string
 		pair1LessPair2 bool
 		pair2LessPair1 bool
+		wantFraction   int
 	}{
-		{"size", true, false},
-		{"name", false, true},
-		{"modtime", false, false},
-		{"size,ascending", true, false},
-		{"name,asc", false, true},
-		{"modtime,ascending", false, false},
-		{"size,descending", false, true},
-		{"name,desc", true, false},
-		{"modtime,descending", true, true},
+		{"size", true, false, -1},
+		{"name", false, true, -1},
+		{"modtime", false, false, -1},
+		{"size,ascending", true, false, -1},
+		{"name,asc", false, true, -1},
+		{"modtime,ascending", false, false, -1},
+		{"size,descending", false, true, -1},
+		{"name,desc", true, false, -1},
+		{"modtime,descending", true, true, -1},
+		{"modtime,mixed", false, false, 50},
+		{"modtime,mixed,30", false, false, 30},
 	} {
 		t.Run(test.orderBy, func(t *testing.T) {
-			less, err := newLess(test.orderBy)
+			less, gotFraction, err := newLess(test.orderBy)
+			assert.Equal(t, test.wantFraction, gotFraction)
 			require.NoError(t, err)
 			require.NotNil(t, less)
 			pair1LessPair2 := less(pair1, pair2)