vfs: synchronize poll updates with shutdown - fixes #9689

vfs/rc.go

@@ -275,6 +275,8 @@ func getStatus(vfs *VFS, in rc.Params) (out rc.Params, err error) {
 	for k, v := range in {
 		return nil, fmt.Errorf("invalid parameter: %s=%s", k, v)
 	}
+	vfs.pollMu.Lock()
+	defer vfs.pollMu.Unlock()
 	return rc.Params{
 		"enabled":   vfs.Opt.PollInterval != 0,
 		"supported": vfs.pollChan != nil,
@@ -332,11 +334,14 @@ func rcPollInterval(ctx context.Context, in rc.Params) (out rc.Params, err error
 	for k, v := range in {
 		return nil, fmt.Errorf("invalid parameter: %s=%s", k, v)
 	}
+	vfs.pollMu.Lock()
 	if vfs.pollChan == nil {
+		vfs.pollMu.Unlock()
 		return nil, errors.New("poll-interval is not supported by this remote")
 	}
 
 	if !intervalPresent {
+		vfs.pollMu.Unlock()
 		return getStatus(vfs, in)
 	}
 	var timeoutHit bool
@@ -351,7 +356,11 @@ func rcPollInterval(ctx context.Context, in rc.Params) (out rc.Params, err error
 		vfs.Opt.PollInterval = fs.Duration(interval)
 	case <-timeoutChan:
 		timeoutHit = true
+	case <-vfs.ctx.Done():
+		vfs.pollMu.Unlock()
+		return nil, errors.New("VFS is shutting down")
 	}
+	vfs.pollMu.Unlock()
 	out, err = getStatus(vfs, in)
 	if out != nil {
 		out["timeout"] = timeoutHit

vfs/rc_test.go

@@ -3,6 +3,7 @@ package vfs
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/rclone/rclone/fs"
 	"github.com/rclone/rclone/fs/rc"
@@ -101,6 +102,75 @@ func TestRcPollInterval(t *testing.T) {
 	// FIXME needs more tests
 }
 
+func TestRcPollIntervalShutdown(t *testing.T) {
+	r := fstest.NewRun(t)
+	features := r.Fremote.Features()
+	originalChangeNotify := features.ChangeNotify
+	t.Cleanup(func() {
+		features.ChangeNotify = originalChangeNotify
+	})
+
+	initialIntervalReceived := make(chan struct{})
+	features.ChangeNotify = func(_ context.Context, _ func(string, fs.EntryType), pollInterval <-chan time.Duration) {
+		go func() {
+			<-pollInterval
+			close(initialIntervalReceived)
+		}()
+	}
+
+	vfs := New(context.Background(), r.Fremote, nil)
+	t.Cleanup(func() {
+		if vfs.inUse.Load() > 0 {
+			vfs.Shutdown()
+		}
+	})
+	<-initialIntervalReceived
+
+	call := rc.Calls.Get("vfs/poll-interval")
+	require.NotNil(t, call)
+	originalInterval := vfs.Opt.PollInterval
+
+	type result struct {
+		out rc.Params
+		err error
+	}
+	resultCh := make(chan result, 1)
+	go func() {
+		out, err := call.Fn(context.Background(), rc.Params{
+			"fs":       fs.ConfigString(r.Fremote),
+			"interval": "1h",
+		})
+		resultCh <- result{out: out, err: err}
+	}()
+
+	select {
+	case got := <-resultCh:
+		t.Fatalf("poll interval update returned before shutdown: out=%v err=%v", got.out, got.err)
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	shutdownDone := make(chan struct{})
+	go func() {
+		vfs.Shutdown()
+		close(shutdownDone)
+	}()
+
+	select {
+	case <-shutdownDone:
+	case <-time.After(time.Second):
+		t.Fatal("VFS shutdown blocked behind poll interval update")
+	}
+
+	select {
+	case got := <-resultCh:
+		require.EqualError(t, got.err, "VFS is shutting down")
+		assert.Nil(t, got.out)
+	case <-time.After(time.Second):
+		t.Fatal("poll interval update did not return after shutdown")
+	}
+	assert.Equal(t, originalInterval, vfs.Opt.PollInterval)
+}
+
 func TestRcList(t *testing.T) {
 	r, vfs, call := rcNewRun(t, "vfs/list")
 	_ = vfs

vfs/vfs.go

@@ -188,6 +188,7 @@ type VFS struct {
 	usageMu     sync.Mutex
 	usageTime   time.Time
 	usage       *fs.Usage
+	pollMu      sync.Mutex
 	pollChan    chan time.Duration
 	inUse       atomic.Int32 // count of number of opens
 }
@@ -410,13 +411,15 @@ func (vfs *VFS) Shutdown() {
 
 	vfs.shutdownCache()
 
+	// Cancel any background go routines
+	vfs.cancel()
+
+	vfs.pollMu.Lock()
 	if vfs.pollChan != nil {
 		close(vfs.pollChan)
 		vfs.pollChan = nil
 	}
-
-	// Cancel any background go routines
-	vfs.cancel()
+	vfs.pollMu.Unlock()
 }
 
 // CleanUp deletes the contents of the on disk cache