vfs: make poll shutdown lifecycle deterministic

vfs/rc.go

@@ -288,6 +288,32 @@ func getStatus(vfs *VFS, in rc.Params) (out rc.Params, err error) {
 	}, nil
 }
 
+func setPollInterval(vfs *VFS, interval, timeout time.Duration) (timeoutHit bool, err error) {
+	vfs.pollMu.Lock()
+	defer vfs.pollMu.Unlock()
+	if vfs.ctx.Err() != nil {
+		return false, errors.New("VFS is shutting down")
+	}
+	if vfs.pollChan == nil {
+		return false, errors.New("poll-interval is not supported by this remote")
+	}
+	var timeoutChan <-chan time.Time
+	if timeout > 0 {
+		timer := time.NewTimer(timeout)
+		defer timer.Stop()
+		timeoutChan = timer.C
+	}
+	select {
+	case vfs.pollChan <- interval:
+		vfs.Opt.PollInterval = fs.Duration(interval)
+	case <-timeoutChan:
+		timeoutHit = true
+	case <-vfs.ctx.Done():
+		return false, errors.New("VFS is shutting down")
+	}
+	return timeoutHit, nil
+}
+
 func init() {
 	rc.Add(rc.Call{
 		Path:  "vfs/poll-interval",
@@ -334,33 +360,19 @@ 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.Lock()
+		supported := vfs.pollChan != nil
 		vfs.pollMu.Unlock()
+		if !supported {
+			return nil, errors.New("poll-interval is not supported by this remote")
+		}
 		return getStatus(vfs, in)
 	}
-	var timeoutHit bool
+	timeoutHit, err := setPollInterval(vfs, interval, timeout)
-	var timeoutChan <-chan time.Time
+	if err != nil {
-	if timeout > 0 {
+		return nil, err
-		timer := time.NewTimer(timeout)
-		defer timer.Stop()
-		timeoutChan = timer.C
 	}
-	select {
-	case vfs.pollChan <- interval:
-		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

@@ -2,6 +2,7 @@ package vfs
 
 import (
 	"context"
+	"runtime"
 	"testing"
 	"time"
 
@@ -110,6 +111,20 @@ func newTestPollVFS(t *testing.T, changeNotify func(context.Context, func(string
 	return r, vfs, call
 }
 
+func waitForPollLock(t *testing.T, vfs *VFS) {
+	t.Helper()
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		if vfs.pollMu.TryLock() {
+			vfs.pollMu.Unlock()
+			runtime.Gosched()
+			continue
+		}
+		return
+	}
+	t.Fatal("poll interval update did not acquire poll lock")
+}
+
 func TestRcPollInterval(t *testing.T) {
 	r, vfs, call := rcNewRun(t, "vfs/poll-interval")
 	_ = vfs
@@ -147,11 +162,7 @@ func TestRcPollIntervalShutdown(t *testing.T) {
 		resultCh <- result{out: out, err: err}
 	}()
 
-	select {
+	waitForPollLock(t, vfs)
-	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() {
@@ -175,6 +186,24 @@ func TestRcPollIntervalShutdown(t *testing.T) {
 	assert.Equal(t, originalInterval, vfs.Opt.PollInterval)
 }
 
+func TestSetPollIntervalAfterShutdown(t *testing.T) {
+	initialIntervalReceived := make(chan struct{})
+	_, vfs, _ := newTestPollVFS(t, func(_ context.Context, _ func(string, fs.EntryType), pollInterval <-chan time.Duration) {
+		go func() {
+			<-pollInterval
+			close(initialIntervalReceived)
+			for range pollInterval {
+			}
+		}()
+	})
+	<-initialIntervalReceived
+	vfs.Shutdown()
+
+	timeoutHit, err := setPollInterval(vfs, time.Hour, 0)
+	require.EqualError(t, err, "VFS is shutting down")
+	assert.False(t, timeoutHit)
+}
+
 func TestRcPollIntervalUpdate(t *testing.T) {
 	intervals := make(chan time.Duration, 2)
 	r, vfs, call := newTestPollVFS(t, func(_ context.Context, _ func(string, fs.EntryType), pollInterval <-chan time.Duration) {