Repository: rclone/rclone
Pull request #7473 (2023-11-30 06:51:58)

oauthutil: add Shutdown method to avoid goroutine leaks
#### What is the purpose of this change?
Before this change, calling the `oauthutil.NewRenew` func may cause goroutine leaks.

This change adds a `Shutdown` method to allow the caller to exit the goroutine to avoid leaks.

#### Was the change discussed in an issue or in the forum before?
None.

#### Checklist

- [x] I have read the [contribution guidelines](https://github.com/rclone/rclone/blob/master/CONTRIBUTING.md#submitting-a-new-feature-or-bug-fix).
- [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 :-)


rkonfj: See also #6692 .
rkonfj: Hi @ncw , can this be merged? 
:)
ncw: Nothing seems to call the Shutdown method?
rkonfj: > Nothing seems to call the Shutdown method?

I checked some backends, amazonclouddrive, box, onedrive, pcloud and other cloud disks call `oauthutil.NewRenew`. 
If this PR is merged, I will create additional PRs (implementing fs.Shutdowner) so that they can release resources.

@ncw May I?
ncw: > > Nothing seems to call the Shutdown method?
> 
> I checked some backends, amazonclouddrive, box, onedrive, pcloud and other cloud disks call `oauthutil.NewRenew`. If this PR is merged, I will create additional PRs (implementing fs.Shutdowner) so that they can release resources.
> 
> @ncw May I?

I see :-)

Can you add them as separate commits to this PR (this could be one commit with all the changes to the backends or one per backend whichever you like)? I like new code to have users before I merge it.

Thank you.
rkonfj: @ncw 
Done, review it again please.

Patch:
diff --git a/backend/amazonclouddrive/amazonclouddrive.go b/backend/amazonclouddrive/amazonclouddrive.go
index 1bc6bcb84..c5a123915 100644
--- a/backend/amazonclouddrive/amazonclouddrive.go
+++ b/backend/amazonclouddrive/amazonclouddrive.go
@@ -1346,6 +1346,12 @@ func (f *Fs) changeNotifyRunner(notifyFunc func(string, fs.EntryType), checkpoin
 	return checkpoint
 }
 
+// Shutdown shutdown the fs
+func (f *Fs) Shutdown(ctx context.Context) error {
+	f.tokenRenewer.Shutdown()
+	return nil
+}
+
 // ID returns the ID of the Object if known, or "" if not
 func (o *Object) ID() string {
 	if o.info.Id == nil {
@@ -1363,6 +1369,7 @@ var (
 	_ fs.DirMover        = (*Fs)(nil)
 	_ fs.DirCacheFlusher = (*Fs)(nil)
 	_ fs.ChangeNotifier  = (*Fs)(nil)
+	_ fs.Shutdowner      = (*Fs)(nil)
 	_ fs.Object          = (*Object)(nil)
 	_ fs.MimeTyper       = &Object{}
 	_ fs.IDer            = &Object{}
diff --git a/backend/box/box.go b/backend/box/box.go
index 26e2cb356..4f38955eb 100644
--- a/backend/box/box.go
+++ b/backend/box/box.go
@@ -1207,6 +1207,12 @@ func (f *Fs) CleanUp(ctx context.Context) (err error) {
 	return err
 }
 
+// Shutdown shutdown the fs
+func (f *Fs) Shutdown(ctx context.Context) error {
+	f.tokenRenewer.Shutdown()
+	return nil
+}
+
 // ChangeNotify calls the passed function with a path that has had changes.
 // If the implementation uses polling, it should adhere to the given interval.
 //
@@ -1719,6 +1725,7 @@ var (
 	_ fs.DirCacheFlusher = (*Fs)(nil)
 	_ fs.PublicLinker    = (*Fs)(nil)
 	_ fs.CleanUpper      = (*Fs)(nil)
+	_ fs.Shutdowner      = (*Fs)(nil)
 	_ fs.Object          = (*Object)(nil)
 	_ fs.IDer            = (*Object)(nil)
 )
diff --git a/backend/dropbox/dropbox.go b/backend/dropbox/dropbox.go
index 1b647af08..0eea08c53 100644
--- a/backend/dropbox/dropbox.go
+++ b/backend/dropbox/dropbox.go
@@ -1231,18 +1231,21 @@ func (f *Fs) About(ctx context.Context) (usage *fs.Usage, err error) {
 		return nil, err
 	}
 	var total uint64
+	var used = q.Used
 	if q.Allocation != nil {
 		if q.Allocation.Individual != nil {
 			total += q.Allocation.Individual.Allocated
 		}
 		if q.Allocation.Team != nil {
 			total += q.Allocation.Team.Allocated
+			// Override used with Team.Used as this includes q.Used already
+			used = q.Allocation.Team.Used
 		}
 	}
 	usage = &fs.Usage{
-		Total: fs.NewUsageValue(int64(total)),          // quota of bytes that can be used
-		Used:  fs.NewUsageValue(int64(q.Used)),         // bytes in use
-		Free:  fs.NewUsageValue(int64(total - q.Used)), // bytes which can be uploaded before reaching the quota
+		Total: fs.NewUsageValue(int64(total)),        // quota of bytes that can be used
+		Used:  fs.NewUsageValue(int64(used)),         // bytes in use
+		Free:  fs.NewUsageValue(int64(total - used)), // bytes which can be uploaded before reaching the quota
 	}
 	return usage, nil
 }
diff --git a/backend/hidrive/hidrive.go b/backend/hidrive/hidrive.go
index 500e6f7cf..6f690f629 100644
--- a/backend/hidrive/hidrive.go
+++ b/backend/hidrive/hidrive.go
@@ -762,6 +762,12 @@ func (f *Fs) DirMove(ctx context.Context, src fs.Fs, srcRemote, dstRemote string
 	return nil
 }
 
+// Shutdown shutdown the fs
+func (f *Fs) Shutdown(ctx context.Context) error {
+	f.tokenRenewer.Shutdown()
+	return nil
+}
+
 // ------------------------------------------------------------
 
 // Fs returns the parent Fs.
@@ -997,6 +1003,7 @@ var (
 	_ fs.Copier         = (*Fs)(nil)
 	_ fs.Mover          = (*Fs)(nil)
 	_ fs.DirMover       = (*Fs)(nil)
+	_ fs.Shutdowner     = (*Fs)(nil)
 	_ fs.Object         = (*Object)(nil)
 	_ fs.IDer           = (*Object)(nil)
 )
diff --git a/backend/jottacloud/jottacloud.go b/backend/jottacloud/jottacloud.go
index e9176f5b0..6fb68b289 100644
--- a/backend/jottacloud/jottacloud.go
+++ b/backend/jottacloud/jottacloud.go
@@ -1680,6 +1680,12 @@ func (f *Fs) CleanUp(ctx context.Context) error {
 	return nil
 }
 
+// Shutdown shutdown the fs
+func (f *Fs) Shutdown(ctx context.Context) error {
+	f.tokenRenewer.Shutdown()
+	return nil
+}
+
 // Hashes returns the supported hash sets.
 func (f *Fs) Hashes() hash.Set {
 	return hash.Set(hash.MD5)
@@ -2104,6 +2110,7 @@ var (
 	_ fs.Abouter      = (*Fs)(nil)
 	_ fs.UserInfoer   = (*Fs)(nil)
 	_ fs.CleanUpper   = (*Fs)(nil)
+	_ fs.Shutdowner   = (*Fs)(nil)
 	_ fs.Object       = (*Object)(nil)
 	_ fs.MimeTyper    = (*Object)(nil)
 	_ fs.Metadataer   = (*Object)(nil)
diff --git a/backend/onedrive/onedrive.go b/backend/onedrive/onedrive.go
index 00781db8d..5d8c68d03 100644
--- a/backend/onedrive/onedrive.go
+++ b/backend/onedrive/onedrive.go
@@ -1377,6 +1377,12 @@ func (f *Fs) ListR(ctx context.Context, dir string, callback fs.ListRCallback) (
 
 }
 
+// Shutdown shutdown the fs
+func (f *Fs) Shutdown(ctx context.Context) error {
+	f.tokenRenewer.Shutdown()
+	return nil
+}
+
 // Creates from the parameters passed in a half finished Object which
 // must have setMetaData called on it
 //
@@ -2755,6 +2761,7 @@ var (
 	_ fs.PublicLinker    = (*Fs)(nil)
 	_ fs.CleanUpper      = (*Fs)(nil)
 	_ fs.ListRer         = (*Fs)(nil)
+	_ fs.Shutdowner      = (*Fs)(nil)
 	_ fs.Object          = (*Object)(nil)
 	_ fs.MimeTyper       = &Object{}
 	_ fs.IDer            = &Object{}
diff --git a/backend/pcloud/pcloud.go b/backend/pcloud/pcloud.go
index 9d6256b1a..64a48c25c 100644
--- a/backend/pcloud/pcloud.go
+++ b/backend/pcloud/pcloud.go
@@ -948,6 +948,12 @@ func (f *Fs) About(ctx context.Context) (usage *fs.Usage, err error) {
 	return usage, nil
 }
 
+// Shutdown shutdown the fs
+func (f *Fs) Shutdown(ctx context.Context) error {
+	f.tokenRenewer.Shutdown()
+	return nil
+}
+
 // Hashes returns the supported hash sets.
 func (f *Fs) Hashes() hash.Set {
 	// EU region supports SHA1 and SHA256 (but rclone doesn't
@@ -1280,6 +1286,7 @@ var (
 	_ fs.DirCacheFlusher = (*Fs)(nil)
 	_ fs.PublicLinker    = (*Fs)(nil)
 	_ fs.Abouter         = (*Fs)(nil)
+	_ fs.Shutdowner      = (*Fs)(nil)
 	_ fs.Object          = (*Object)(nil)
 	_ fs.IDer            = (*Object)(nil)
 )
diff --git a/backend/premiumizeme/premiumizeme.go b/backend/premiumizeme/premiumizeme.go
index 4adaa5af2..ef54aac9f 100644
--- a/backend/premiumizeme/premiumizeme.go
+++ b/backend/premiumizeme/premiumizeme.go
@@ -770,6 +770,12 @@ func (f *Fs) PublicLink(ctx context.Context, remote string, expire fs.Duration,
 	return o.(*Object).url, nil
 }
 
+// Shutdown shutdown the fs
+func (f *Fs) Shutdown(ctx context.Context) error {
+	f.tokenRenewer.Shutdown()
+	return nil
+}
+
 // About gets quota information
 func (f *Fs) About(ctx context.Context) (usage *fs.Usage, err error) {
 	var resp *http.Response
@@ -1110,6 +1116,7 @@ var (
 	_ fs.DirCacheFlusher = (*Fs)(nil)
 	_ fs.Abouter         = (*Fs)(nil)
 	_ fs.PublicLinker    = (*Fs)(nil)
+	_ fs.Shutdowner      = (*Fs)(nil)
 	_ fs.Object          = (*Object)(nil)
 	_ fs.MimeTyper       = (*Object)(nil)
 	_ fs.IDer            = (*Object)(nil)
diff --git a/backend/s3/s3.go b/backend/s3/s3.go
index 979423883..d7ca47a77 100644
--- a/backend/s3/s3.go
+++ b/backend/s3/s3.go
@@ -2219,6 +2219,13 @@ If it is set then rclone will use v2 authentication.
 Use this only if v4 signatures don't work, e.g. pre Jewel/v10 CEPH.`,
 			Default:  false,
 			Advanced: true,
+		}, {
+			Name: "use_dual_stack",
+			Help: `If true use AWS S3 dual-stack endpoint (IPv6 support).
+
+See [AWS Docs on Dualstack Endpoints](https://docs.aws.amazon.com/AmazonS3/latest/userguide/dual-stack-endpoints.html)`,
+			Default:  false,
+			Advanced: true,
 		}, {
 			Name:     "use_accelerate_endpoint",
 			Provider: "AWS",
@@ -2628,6 +2635,7 @@ type Options struct {
 	Region                string               `config:"region"`
 	Endpoint              string               `config:"endpoint"`
 	STSEndpoint           string               `config:"sts_endpoint"`
+	UseDualStack          bool                 `config:"use_dual_stack"`
 	LocationConstraint    string               `config:"location_constraint"`
 	ACL                   string               `config:"acl"`
 	BucketACL             string               `config:"bucket_acl"`
@@ -2956,6 +2964,9 @@ func s3Connection(ctx context.Context, opt *Options, client *http.Client) (*s3.S
 		r.addService("sts", opt.STSEndpoint)
 		awsConfig.WithEndpointResolver(r)
 	}
+	if opt.UseDualStack {
+		awsConfig.UseDualStackEndpoint = endpoints.DualStackEndpointStateEnabled
+	}
 
 	// awsConfig.WithLogLevel(aws.LogDebugWithSigning)
 	awsSessionOpts := session.Options{
diff --git a/backend/sharefile/sharefile.go b/backend/sharefile/sharefile.go
index 92e618010..a135f8705 100644
--- a/backend/sharefile/sharefile.go
+++ b/backend/sharefile/sharefile.go
@@ -1176,6 +1176,12 @@ func (f *Fs) DirCacheFlush() {
 	f.dirCache.ResetRoot()
 }
 
+// Shutdown shutdown the fs
+func (f *Fs) Shutdown(ctx context.Context) error {
+	f.tokenRenewer.Shutdown()
+	return nil
+}
+
 // Hashes returns the supported hash sets.
 func (f *Fs) Hashes() hash.Set {
 	return hash.Set(hash.MD5)
@@ -1466,6 +1472,7 @@ var (
 	_ fs.Copier   = (*Fs)(nil)
 	// _ fs.PutStreamer     = (*Fs)(nil)
 	_ fs.DirCacheFlusher = (*Fs)(nil)
+	_ fs.Shutdowner      = (*Fs)(nil)
 	_ fs.Object          = (*Object)(nil)
 	_ fs.IDer            = (*Object)(nil)
 )
diff --git a/bin/.ignore-emails b/bin/.ignore-emails
index 14cbc62fc..b45311099 100644
--- a/bin/.ignore-emails
+++ b/bin/.ignore-emails
@@ -10,3 +10,4 @@
 <stoesser@yay-digital.de>
 <services+github@simjo.st>
 <seb•ɑƬ•chezwam•ɖɵʈ•org>
+<allllaboutyou@gmail.com>
\ No newline at end of file
diff --git a/docs/content/authors.md b/docs/content/authors.md
index 88f7cf1ce..66ff5a961 100644
--- a/docs/content/authors.md
+++ b/docs/content/authors.md
@@ -57,7 +57,7 @@ put them back in again.` >}}
   * Scott McGillivray <scott.mcgillivray@gmail.com>
   * Bjørn Erik Pedersen <bjorn.erik.pedersen@gmail.com>
   * Lukas Loesche <lukas@mesosphere.io>
-  * emyarod <allllaboutyou@gmail.com>
+  * emyarod <emyarod@users.noreply.github.com>
   * T.C. Ferguson <tcf909@gmail.com>
   * Brandur <brandur@mutelight.org>
   * Dario Giovannetti <dev@dariogiovannetti.net>
@@ -807,3 +807,6 @@ put them back in again.` >}}
   * Abhinav Dhiman <8640877+ahnv@users.noreply.github.com>
   * halms <7513146+halms@users.noreply.github.com>
   * ben-ba <benjamin.brauner@gmx.de>
+  * Eli Orzitzer <e_orz@yahoo.com>
+  * Anthony Metzidis <anthony.metzidis@gmail.com>
+  * emyarod <afw5059@gmail.com>
diff --git a/docs/content/docs.md b/docs/content/docs.md
index fc40963f0..a15cc5188 100644
--- a/docs/content/docs.md
+++ b/docs/content/docs.md
@@ -1614,7 +1614,7 @@ json.dump(o, sys.stdout, indent="\t")
 ```
 
 You can find this example (slightly expanded) in the rclone source code at
-[bin/test_metadata_mapper.py](https://github.com/rclone/rclone/blob/master/test_metadata_mapper.py).
+[bin/test_metadata_mapper.py](https://github.com/rclone/rclone/blob/master/bin/test_metadata_mapper.py).
 
 If you want to see the input to the metadata mapper and the output
 returned from it in the log you can use `-vv --dump mapper`.
diff --git a/lib/oauthutil/renew.go b/lib/oauthutil/renew.go
index 7f98491e8..e3652ee9c 100644
--- a/lib/oauthutil/renew.go
+++ b/lib/oauthutil/renew.go
@@ -1,6 +1,7 @@
 package oauthutil
 
 import (
+	"sync"
 	"sync/atomic"
 
 	"github.com/rclone/rclone/fs"
@@ -8,10 +9,12 @@ import (
 
 // Renew allows tokens to be renewed on expiry if uploads are in progress.
 type Renew struct {
-	name    string       // name to use in logs
-	ts      *TokenSource // token source that needs renewing
-	uploads atomic.Int32 // number of uploads in progress
-	run     func() error // a transaction to run to renew the token on
+	name     string       // name to use in logs
+	ts       *TokenSource // token source that needs renewing
+	uploads  atomic.Int32 // number of uploads in progress
+	run      func() error // a transaction to run to renew the token on
+	done     chan any     // channel to end the go routine
+	shutdown sync.Once
 }
 
 // NewRenew creates a new Renew struct and starts a background process
@@ -24,6 +27,7 @@ func NewRenew(name string, ts *TokenSource, run func() error) *Renew {
 		name: name,
 		ts:   ts,
 		run:  run,
+		done: make(chan any),
 	}
 	go r.renewOnExpiry()
 	return r
@@ -36,7 +40,11 @@ func NewRenew(name string, ts *TokenSource, run func() error) *Renew {
 func (r *Renew) renewOnExpiry() {
 	expiry := r.ts.OnExpiry()
 	for {
-		<-expiry
+		select {
+		case <-expiry:
+		case <-r.done:
+			return
+		}
 		uploads := r.uploads.Load()
 		if uploads != 0 {
 			fs.Debugf(r.name, "Token expired - %d uploads in progress - refreshing", uploads)
@@ -72,3 +80,15 @@ func (r *Renew) Invalidate() {
 func (r *Renew) Expire() error {
 	return r.ts.Expire()
 }
+
+// Shutdown stops the timer and no more renewal will take place.
+func (r *Renew) Shutdown() {
+	if r == nil {
+		return
+	}
+	// closing a channel can only be done once
+	r.shutdown.Do(func() {
+		r.ts.expiryTimer.Stop()
+		close(r.done)
+	})
+}
diff --git a/vfs/vfsflags/vfsflags.go b/vfs/vfsflags/vfsflags.go
index 2f1adafde..83b593d29 100644
--- a/vfs/vfsflags/vfsflags.go
+++ b/vfs/vfsflags/vfsflags.go
@@ -22,7 +22,7 @@ func AddFlags(flagSet *pflag.FlagSet) {
 	flags.BoolVarP(flagSet, &Opt.NoChecksum, "no-checksum", "", Opt.NoChecksum, "Don't compare checksums on up/download", "VFS")
 	flags.BoolVarP(flagSet, &Opt.NoSeek, "no-seek", "", Opt.NoSeek, "Don't allow seeking in files", "VFS")
 	flags.DurationVarP(flagSet, &Opt.DirCacheTime, "dir-cache-time", "", Opt.DirCacheTime, "Time to cache directory entries for", "VFS")
-	flags.BoolVarP(flagSet, &Opt.Refresh, "vfs-refresh", "", Opt.Refresh, "Refreshes the directory cache recursively on start", "VFS")
+	flags.BoolVarP(flagSet, &Opt.Refresh, "vfs-refresh", "", Opt.Refresh, "Refreshes the directory cache recursively in the background on start", "VFS")
 	flags.DurationVarP(flagSet, &Opt.PollInterval, "poll-interval", "", Opt.PollInterval, "Time to wait between polling for changes, must be smaller than dir-cache-time and only on supported remotes (set 0 to disable)", "VFS")
 	flags.BoolVarP(flagSet, &Opt.ReadOnly, "read-only", "", Opt.ReadOnly, "Only allow read-only access", "VFS")
 	flags.FVarP(flagSet, &Opt.CacheMode, "vfs-cache-mode", "", "Cache mode off|minimal|writes|full", "VFS")


Test patch:
diff --git a/backend/s3/s3_test.go b/backend/s3/s3_test.go
index 3415bebf2..7e6e55360 100644
--- a/backend/s3/s3_test.go
+++ b/backend/s3/s3_test.go
@@ -2,6 +2,9 @@
 package s3
 
 import (
+	"context"
+	"net/http"
+	"strings"
 	"testing"
 
 	"github.com/rclone/rclone/fs"
@@ -9,6 +12,13 @@ import (
 	"github.com/rclone/rclone/fstest/fstests"
 )
 
+func SetupS3Test(t *testing.T) (context.Context, *Options, *http.Client) {
+	ctx, opt := context.Background(), new(Options)
+	opt.Provider = "AWS"
+	client := getClient(ctx, opt)
+	return ctx, opt, client
+}
+
 // TestIntegration runs integration tests against the remote
 func TestIntegration(t *testing.T) {
 	fstests.Run(t, &fstests.Opt{
@@ -39,6 +49,28 @@ func TestIntegration2(t *testing.T) {
 	})
 }
 
+func TestAWSDualStackOption(t *testing.T) {
+	{
+		// test enabled
+		ctx, opt, client := SetupS3Test(t)
+		opt.UseDualStack = true
+		s3Conn, _, _ := s3Connection(ctx, opt, client)
+		if !strings.Contains(s3Conn.Endpoint, "dualstack") {
+			t.Errorf("dualstack failed got: %s, wanted: dualstack", s3Conn.Endpoint)
+			t.Fail()
+		}
+	}
+	{
+		// test default case
+		ctx, opt, client := SetupS3Test(t)
+		s3Conn, _, _ := s3Connection(ctx, opt, client)
+		if strings.Contains(s3Conn.Endpoint, "dualstack") {
+			t.Errorf("dualstack failed got: %s, NOT wanted: dualstack", s3Conn.Endpoint)
+			t.Fail()
+		}
+	}
+}
+
 func (f *Fs) SetUploadChunkSize(cs fs.SizeSuffix) (fs.SizeSuffix, error) {
 	return f.setUploadChunkSize(cs)
 }