Merge branch 'master' of github.com:rclone/rclone into feature/storj

Author: calebcase

VERSION

@@ -1 +1 @@
-v1.50.1
+v1.50.2

backend/cache/storage_persistent.go

@@ -16,7 +16,7 @@ import (
 	"sync"
 	"time"
 
-	bolt "github.com/coreos/bbolt"
+	bolt "github.com/etcd-io/bbolt"
 	"github.com/pkg/errors"
 	"github.com/rclone/rclone/fs"
 	"github.com/rclone/rclone/fs/walk"

backend/drive/drive.go

@@ -1021,11 +1021,11 @@ func NewFs(name, path string, m configmap.Mapper) (fs.Fs, error) {
 	}
 
 	// set root folder for a team drive or query the user root folder
-	if f.isTeamDrive {
-		f.rootFolderID = f.opt.TeamDriveID
-	} else if opt.RootFolderID != "" {
+	if opt.RootFolderID != "" {
 		// override root folder if set or cached in the config
 		f.rootFolderID = opt.RootFolderID
+	} else if f.isTeamDrive {
+		f.rootFolderID = f.opt.TeamDriveID
 	} else {
 		// Look up the root ID and cache it in the config
 		rootID, err := f.getRootID()

backend/koofr/koofr.go

@@ -16,6 +16,7 @@ import (
 	"github.com/rclone/rclone/fs/config/configstruct"
 	"github.com/rclone/rclone/fs/config/obscure"
 	"github.com/rclone/rclone/fs/encodings"
+	"github.com/rclone/rclone/fs/fshttp"
 	"github.com/rclone/rclone/fs/hash"
 
 	httpclient "github.com/koofr/go-httpclient"
@@ -259,7 +260,9 @@ func NewFs(name, root string, m configmap.Mapper) (ff fs.Fs, err error) {
 	if err != nil {
 		return nil, err
 	}
-	client := koofrclient.NewKoofrClient(opt.Endpoint, false)
+	httpClient := httpclient.New()
+	httpClient.Client = fshttp.NewClient(fs.Config)
+	client := koofrclient.NewKoofrClientWithHTTPClient(opt.Endpoint, httpClient)
 	basicAuth := fmt.Sprintf("Basic %s",
 		base64.StdEncoding.EncodeToString([]byte(opt.User+":"+pass)))
 	client.HTTPClient.Headers.Set("Authorization", basicAuth)

backend/local/local.go

@@ -350,7 +350,7 @@ func (f *Fs) List(ctx context.Context, dir string) (entries fs.DirEntries, err e
 		err = errors.Wrapf(err, "failed to open directory %q", dir)
 		fs.Errorf(dir, "%v", err)
 		if isPerm {
-			accounting.Stats(ctx).Error(fserrors.NoRetryError(err))
+			_ = accounting.Stats(ctx).Error(fserrors.NoRetryError(err))
 			err = nil // ignore error but fail sync
 		}
 		return nil, err
@@ -386,7 +386,7 @@ func (f *Fs) List(ctx context.Context, dir string) (entries fs.DirEntries, err e
 					if fierr != nil {
 						err = errors.Wrapf(err, "failed to read directory %q", namepath)
 						fs.Errorf(dir, "%v", fierr)
-						accounting.Stats(ctx).Error(fserrors.NoRetryError(fierr)) // fail the sync
+						_ = accounting.Stats(ctx).Error(fserrors.NoRetryError(fierr)) // fail the sync
 						continue
 					}
 					fis = append(fis, fi)
@@ -409,7 +409,7 @@ func (f *Fs) List(ctx context.Context, dir string) (entries fs.DirEntries, err e
 					// Skip bad symlinks
 					err = fserrors.NoRetryError(errors.Wrap(err, "symlink"))
 					fs.Errorf(newRemote, "Listing error: %v", err)
-					accounting.Stats(ctx).Error(err)
+					err = accounting.Stats(ctx).Error(err)
 					continue
 				}
 				if err != nil {

backend/qingstor/qingstor.go

@@ -269,7 +269,7 @@ func qsServiceConnection(opt *Options) (*qs.Service, error) {
 	cf.Protocol = protocol
 	cf.Host = host
 	cf.Port = port
-	cf.ConnectionRetries = opt.ConnectionRetries
+	// unsupported in v3.1: cf.ConnectionRetries = opt.ConnectionRetries
 	cf.Connection = fshttp.NewClient(fs.Config)
 
 	return qs.Init(cf)

backend/sftp/sftp.go

@@ -29,15 +29,17 @@ import (
 	"github.com/rclone/rclone/fs/fshttp"
 	"github.com/rclone/rclone/fs/hash"
 	"github.com/rclone/rclone/lib/env"
+	"github.com/rclone/rclone/lib/pacer"
 	"github.com/rclone/rclone/lib/readers"
 	sshagent "github.com/xanzy/ssh-agent"
 	"golang.org/x/crypto/ssh"
-	"golang.org/x/time/rate"
 )
 
 const (
-	connectionsPerSecond    = 10 // don't make more than this many ssh connections/s
 	hashCommandNotSupported = "none"
+	minSleep                = 100 * time.Millisecond
+	maxSleep                = 2 * time.Second
+	decayConstant           = 2 // bigger for slower decay, exponential
 )
 
 var (
@@ -190,7 +192,7 @@ type Fs struct {
 	cachedHashes *hash.Set
 	poolMu       sync.Mutex
 	pool         []*conn
-	connLimit    *rate.Limiter // for limiting number of connections per second
+	pacer        *fs.Pacer // pacer for operations
 }
 
 // Object is a remote SFTP file that has been stat'd (so it exists, but is not necessarily open for reading)
@@ -270,10 +272,6 @@ func (c *conn) closed() error {
 // Open a new connection to the SFTP server.
 func (f *Fs) sftpConnection() (c *conn, err error) {
 	// Rate limit rate of new connections
-	err = f.connLimit.Wait(context.Background())
-	if err != nil {
-		return nil, errors.Wrap(err, "limiter failed in connect")
-	}
 	c = &conn{
 		err: make(chan error, 1),
 	}
@@ -307,7 +305,14 @@ func (f *Fs) getSftpConnection() (c *conn, err error) {
 	if c != nil {
 		return c, nil
 	}
-	return f.sftpConnection()
+	err = f.pacer.Call(func() (bool, error) {
+		c, err = f.sftpConnection()
+		if err != nil {
+			return true, err
+		}
+		return false, nil
+	})
+	return c, err
 }
 
 // Return an SFTP connection to the pool
@@ -465,7 +470,7 @@ func NewFsWithConnection(ctx context.Context, name string, root string, m config
 		config:    sshConfig,
 		url:       "sftp://" + opt.User + "@" + opt.Host + ":" + opt.Port + "/" + root,
 		mkdirLock: newStringLock(),
-		connLimit: rate.NewLimiter(rate.Limit(connectionsPerSecond), 1),
+		pacer:     fs.NewPacer(pacer.NewDefault(pacer.MinSleep(minSleep), pacer.MaxSleep(maxSleep), pacer.DecayConstant(decayConstant))),
 	}
 	f.features = (&fs.Features{
 		CanHaveEmptyDirectories: true,

backend/swift/swift.go

@@ -530,10 +530,10 @@ type listFn func(remote string, object *swift.Object, isDirectory bool) error
 //
 // Set recurse to read sub directories
 func (f *Fs) listContainerRoot(container, directory, prefix string, addContainer bool, recurse bool, fn listFn) error {
-	if prefix != "" {
+	if prefix != "" && !strings.HasSuffix(prefix, "/") {
 		prefix += "/"
 	}
-	if directory != "" {
+	if directory != "" && !strings.HasSuffix(directory, "/") {
 		directory += "/"
 	}
 	// Options for ObjectsWalk
@@ -952,6 +952,18 @@ func (o *Object) isStaticLargeObject() (bool, error) {
 	return o.hasHeader("X-Static-Large-Object")
 }
 
+func (o *Object) isInContainerVersioning() (bool, error) {
+	_, headers, err := o.fs.c.Container(o.fs.root)
+	if err != nil {
+		return false, err
+	}
+	xHistoryLocation := headers["X-History-Location"]
+	if len(xHistoryLocation) > 0 {
+		return true, nil
+	}
+	return false, nil
+}
+
 // Size returns the size of an object in bytes
 func (o *Object) Size() int64 {
 	return o.size
@@ -1083,9 +1095,8 @@ func min(x, y int64) int64 {
 //
 // if except is passed in then segments with that prefix won't be deleted
 func (o *Object) removeSegments(except string) error {
-	container, containerPath := o.split()
-	segmentsContainer := container + "_segments"
-	err := o.fs.listContainerRoot(segmentsContainer, containerPath, "", false, true, func(remote string, object *swift.Object, isDirectory bool) error {
+	segmentsContainer, prefix, err := o.getSegmentsDlo()
+	err = o.fs.listContainerRoot(segmentsContainer, prefix, "", false, true, func(remote string, object *swift.Object, isDirectory bool) error {
 		if isDirectory {
 			return nil
 		}
@@ -1114,6 +1125,19 @@ func (o *Object) removeSegments(except string) error {
 	return nil
 }
 
+func (o *Object) getSegmentsDlo() (segmentsContainer string, prefix string, err error) {
+	if err = o.readMetaData(); err != nil {
+		return
+	}
+	dirManifest := o.headers["X-Object-Manifest"]
+	delimiter := strings.Index(dirManifest, "/")
+	if len(dirManifest) == 0 || delimiter < 0 {
+		err = errors.New("Missing or wrong structure of manifest of Dynamic large object")
+		return
+	}
+	return dirManifest[:delimiter], dirManifest[delimiter+1:], nil
+}
+
 // urlEncode encodes a string so that it is a valid URL
 //
 // We don't use any of Go's standard methods as we need `/` not
@@ -1300,12 +1324,9 @@ func (o *Object) Update(ctx context.Context, in io.Reader, src fs.ObjectInfo, op
 }
 
 // Remove an object
-func (o *Object) Remove(ctx context.Context) error {
+func (o *Object) Remove(ctx context.Context) (err error) {
 	container, containerPath := o.split()
-	isDynamicLargeObject, err := o.isDynamicLargeObject()
-	if err != nil {
-		return err
-	}
+
 	// Remove file/manifest first
 	err = o.fs.pacer.Call(func() (bool, error) {
 		err = o.fs.c.ObjectDelete(container, containerPath)
@@ -1314,12 +1335,22 @@ func (o *Object) Remove(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
+	isDynamicLargeObject, err := o.isDynamicLargeObject()
+	if err != nil {
+		return err
+	}
 	// ...then segments if required
 	if isDynamicLargeObject {
-		err = o.removeSegments("")
+		isInContainerVersioning, err := o.isInContainerVersioning()
 		if err != nil {
 			return err
 		}
+		if !isInContainerVersioning {
+			err = o.removeSegments("")
+			if err != nil {
+				return err
+			}
+		}
 	}
 	return nil
 }

cmd/cmd.go

@@ -82,7 +82,7 @@ func ShowVersion() {
 func NewFsFile(remote string) (fs.Fs, string) {
 	_, _, fsPath, err := fs.ParseRemote(remote)
 	if err != nil {
-		fs.CountError(err)
+		err = fs.CountError(err)
 		log.Fatalf("Failed to create file system for %q: %v", remote, err)
 	}
 	f, err := cache.Get(remote)
@@ -92,7 +92,7 @@ func NewFsFile(remote string) (fs.Fs, string) {
 	case nil:
 		return f, ""
 	default:
-		fs.CountError(err)
+		err = fs.CountError(err)
 		log.Fatalf("Failed to create file system for %q: %v", remote, err)
 	}
 	return nil, ""
@@ -107,13 +107,13 @@ func newFsFileAddFilter(remote string) (fs.Fs, string) {
 	if fileName != "" {
 		if !filter.Active.InActive() {
 			err := errors.Errorf("Can't limit to single files when using filters: %v", remote)
-			fs.CountError(err)
+			err = fs.CountError(err)
 			log.Fatalf(err.Error())
 		}
 		// Limit transfers to this file
 		err := filter.Active.AddFile(fileName)
 		if err != nil {
-			fs.CountError(err)
+			err = fs.CountError(err)
 			log.Fatalf("Failed to limit to single file %q: %v", remote, err)
 		}
 	}
@@ -135,7 +135,7 @@ func NewFsSrc(args []string) fs.Fs {
 func newFsDir(remote string) fs.Fs {
 	f, err := cache.Get(remote)
 	if err != nil {
-		fs.CountError(err)
+		err = fs.CountError(err)
 		log.Fatalf("Failed to create file system for %q: %v", remote, err)
 	}
 	return f
@@ -189,11 +189,11 @@ func NewFsSrcDstFiles(args []string) (fsrc fs.Fs, srcFileName string, fdst fs.Fs
 	fdst, err := cache.Get(dstRemote)
 	switch err {
 	case fs.ErrorIsFile:
-		fs.CountError(err)
+		_ = fs.CountError(err)
 		log.Fatalf("Source doesn't exist or is a directory and destination is a file")
 	case nil:
 	default:
-		fs.CountError(err)
+		_ = fs.CountError(err)
 		log.Fatalf("Failed to create file system for destination %q: %v", dstRemote, err)
 	}
 	return
@@ -239,7 +239,7 @@ func Run(Retry bool, showStats bool, cmd *cobra.Command, f func() error) {
 	SigInfoHandler()
 	for try := 1; try <= *retries; try++ {
 		err = f()
-		fs.CountError(err)
+		err = fs.CountError(err)
 		lastErr := accounting.GlobalStats().GetLastError()
 		if err == nil {
 			err = lastErr
@@ -386,12 +386,12 @@ func initConfig() {
 		fs.Infof(nil, "Creating CPU profile %q\n", *cpuProfile)
 		f, err := os.Create(*cpuProfile)
 		if err != nil {
-			fs.CountError(err)
+			err = fs.CountError(err)
 			log.Fatal(err)
 		}
 		err = pprof.StartCPUProfile(f)
 		if err != nil {
-			fs.CountError(err)
+			err = fs.CountError(err)
 			log.Fatal(err)
 		}
 		atexit.Register(func() {
@@ -405,17 +405,17 @@ func initConfig() {
 			fs.Infof(nil, "Saving Memory profile %q\n", *memProfile)
 			f, err := os.Create(*memProfile)
 			if err != nil {
-				fs.CountError(err)
+				err = fs.CountError(err)
 				log.Fatal(err)
 			}
 			err = pprof.WriteHeapProfile(f)
 			if err != nil {
-				fs.CountError(err)
+				err = fs.CountError(err)
 				log.Fatal(err)
 			}
 			err = f.Close()
 			if err != nil {
-				fs.CountError(err)
+				err = fs.CountError(err)
 				log.Fatal(err)
 			}
 		})

cmd/cryptcheck/cryptcheck.go

@@ -88,7 +88,7 @@ func cryptCheck(ctx context.Context, fdst, fsrc fs.Fs) error {
 		underlyingDst := cryptDst.UnWrap()
 		underlyingHash, err := underlyingDst.Hash(ctx, hashType)
 		if err != nil {
-			fs.CountError(err)
+			err = fs.CountError(err)
 			fs.Errorf(dst, "Error reading hash from underlying %v: %v", underlyingDst, err)
 			return true, false
 		}
@@ -97,7 +97,7 @@ func cryptCheck(ctx context.Context, fdst, fsrc fs.Fs) error {
 		}
 		cryptHash, err := fcrypt.ComputeHash(ctx, cryptDst, src, hashType)
 		if err != nil {
-			fs.CountError(err)
+			err = fs.CountError(err)
 			fs.Errorf(dst, "Error computing hash: %v", err)
 			return true, false
 		}
@@ -106,7 +106,7 @@ func cryptCheck(ctx context.Context, fdst, fsrc fs.Fs) error {
 		}
 		if cryptHash != underlyingHash {
 			err = errors.Errorf("hashes differ (%s:%s) %q vs (%s:%s) %q", fdst.Name(), fdst.Root(), cryptHash, fsrc.Name(), fsrc.Root(), underlyingHash)
-			fs.CountError(err)
+			err = fs.CountError(err)
 			fs.Errorf(src, err.Error())
 			return true, false
 		}

cmd/serve/dlna/dlna_util.go

@@ -214,7 +214,7 @@ func withHeader(name string, value string, next http.Handler) http.Handler {
 
 // serveError returns an http.StatusInternalServerError and logs the error
 func serveError(what interface{}, w http.ResponseWriter, text string, err error) {
-	fs.CountError(err)
+	err = fs.CountError(err)
 	fs.Errorf(what, "%s: %v", text, err)
 	http.Error(w, text+".", http.StatusInternalServerError)
 }

cmd/serve/httplib/serve/dir.go

@@ -68,7 +68,7 @@ func (d *Directory) AddEntry(remote string, isDir bool) {
 
 // Error logs the error and if a ResponseWriter is given it writes a http.StatusInternalServerError
 func Error(what interface{}, w http.ResponseWriter, text string, err error) {
-	fs.CountError(err)
+	err = fs.CountError(err)
 	fs.Errorf(what, "%s: %v", text, err)
 	if w != nil {
 		http.Error(w, text+".", http.StatusInternalServerError)