Merge ac602eecc20913a1c968d3b737ba4c8afb719f54 into 0b9671313b14ffe839ecbd7dd2ae5ac7f6f05db8

backend/compress/compress.go

@@ -274,6 +274,21 @@ func isMetadataFile(filename string) bool {
 	return strings.HasSuffix(filename, metaFileExt)
 }
 
+// Checks whether a file is a compressed file
+func isCompressedFile(filename string) bool {
+	return strings.HasSuffix(filename, gzFileExt)
+}
+
+// Checks whether a file is an uncompressed file
+func isUncompressedFile(filename string) bool {
+	return strings.HasSuffix(filename, uncompressedFileExt)
+}
+
+// Checks whether a path is a file, all of which must have one of the special file extensions
+func isFile(filename string) bool {
+	return isMetadataFile(filename) || isCompressedFile(filename) || isUncompressedFile(filename)
+}
+
 // Checks whether a file is a metadata file and returns the original
 // file name and a flag indicating whether it was a metadata file or
 // not.
@@ -1053,6 +1068,17 @@ func (f *Fs) ChangeNotify(ctx context.Context, notifyFunc func(string, fs.EntryT
 			wrappedPath    string
 			isMetadataFile bool
 		)
+
+		// Uncertain entry types can be resolved by the fact that all files in the compressed
+		// backend have a special file extension
+		if entryType == fs.EntryUncertain {
+			if isFile(path) {
+				entryType = fs.EntryObject
+			} else {
+				entryType = fs.EntryDirectory
+			}
+		}
+
 		switch entryType {
 		case fs.EntryDirectory:
 			wrappedPath = path

backend/crypt/crypt.go

@@ -907,6 +907,17 @@ func (f *Fs) ChangeNotify(ctx context.Context, notifyFunc func(string, fs.EntryT
 			decrypted, err = f.cipher.DecryptDirName(path)
 		case fs.EntryObject:
 			decrypted, err = f.cipher.DecryptFileName(path)
+		case fs.EntryUncertain:
+			if f.opt.FilenameEncryption == "on" && !f.opt.DirectoryNameEncryption {
+				// Uncertain entry types cannot be handled in this case, as not only is the entry
+				// type uncertain, but also whether or not the name is encrypted
+				fs.Errorf(path, "crypt ChangeNotify: ignoring EntryType %d as file names are encrypted but directory names are not", fs.EntryUncertain)
+				return
+			}
+			decrypted, err = f.cipher.DecryptFileName(path)
+			if err == ErrorNotAnEncryptedFile {
+				decrypted, err = f.cipher.DecryptDirName(path)
+			}
 		default:
 			fs.Errorf(path, "crypt ChangeNotify: ignoring unknown EntryType %d", entryType)
 			return

backend/local/changenotify_other.go

@@ -0,0 +1,251 @@
+//go:build !windows
+
+package local
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/fsnotify/fsnotify"
+	"github.com/moby/sys/mountinfo"
+	"github.com/rclone/rclone/fs"
+	"github.com/rclone/rclone/fs/walk"
+)
+
+// ChangeNotify calls the passed function with a path that has had changes.
+// Close the returned channel to stop being notified.
+func (f *Fs) ChangeNotify(ctx context.Context, notifyFunc func(string, fs.EntryType), pollIntervalChan <-chan time.Duration) {
+	// Will not work with an NFS mounted filesystem, error in this case
+	infos, err := mountinfo.GetMounts(mountinfo.ParentsFilter(f.root))
+	if err == nil {
+		for i := 0; i < len(infos); i++ {
+			if infos[i].FSType == "nfs" {
+				fs.Error(f, "ChangeNotify does not support NFS mounts")
+				return
+			}
+		}
+	}
+
+	// Create new watcher
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		fs.Errorf(f, "Failed to create watcher: %s", err)
+		return
+	}
+
+	// Files and directories changed in the last poll window, mapped to the
+	// time at which notification of the change was received.
+	changed := make(map[string]time.Time)
+
+	// Channel to handle new paths. Buffered ensures filesystem events keep
+	// being consumed.
+	watchChan := make(chan string)
+
+	// Channel to synchronize with the watch goroutine
+	replyChan := make(chan bool)
+
+	// Start goroutine to handle filesystem events
+	go func() {
+		// Polling is imitated by accumulating events between ticks. While
+		// notifyFunc() could be called immediately on each filesystem event,
+		// accumulating turns out to have some advantages in accurately keeping
+		// track of entry types (i.e. file or directory), under the
+		// interpretation that the notifications sent at each tick are a diff of
+		// the state of the filesystem at that tick compared to the previous. It
+		// is also assumed by some tests.
+		var ticker *time.Ticker
+		var tickerC <-chan time.Time
+
+	loop:
+		for {
+			select {
+			case pollInterval, ok := <-pollIntervalChan:
+				// Update ticker
+				if !ok {
+					if ticker != nil {
+						ticker.Stop()
+					}
+					break loop
+				}
+				if ticker != nil {
+					ticker.Stop()
+					ticker, tickerC = nil, nil
+				}
+				if pollInterval != 0 {
+					ticker = time.NewTicker(pollInterval)
+					tickerC = ticker.C
+				}
+			case <-tickerC:
+				// Notify for all paths that have changed since the last sync, and
+				// which were changed at least 1/10 of a second (1e8 nanoseconds)
+				// ago. The lag is for de-duping purposes during long writes, which
+				// can consist of multiple write notifications in quick succession.
+				cutoff := time.Now().Add(-1e8)
+				for entryPath, entryTime := range changed {
+					if entryTime.Before(cutoff) {
+						notifyFunc(filepath.ToSlash(entryPath), fs.EntryUncertain)
+						delete(changed, entryPath)
+					}
+				}
+			case event, ok := <-watcher.Events:
+				if !ok {
+					break loop
+				}
+				if event.Has(fsnotify.Create) {
+					fs.Debugf(f, "Create: %s", event.Name)
+				}
+				if event.Has(fsnotify.Remove) {
+					fs.Debugf(f, "Remove: %s", event.Name)
+				}
+				if event.Has(fsnotify.Rename) {
+					fs.Debugf(f, "Rename: %s", event.Name)
+				}
+				if event.Has(fsnotify.Write) {
+					fs.Debugf(f, "Write: %s", event.Name)
+				}
+				if event.Has(fsnotify.Chmod) {
+					fs.Debugf(f, "Chmod: %s", event.Name)
+				}
+
+				if event.Has(fsnotify.Create) {
+					fs.Debugf(f, "Create: %s", event.Name)
+					watchChan <- event.Name
+					<-replyChan // implies mutex on 'changed'
+				} else {
+					entryPath, _ := filepath.Rel(f.root, event.Name)
+					changed[entryPath] = time.Now()
+
+					// Internally, fsnotify stops watching directories that are removed
+					// or renamed, so it is not necessary to make updates to the watch
+					// list.
+				}
+			case err, ok := <-watcher.Errors:
+				if !ok {
+					break loop
+				}
+				fs.Errorf(f, "Error: %s", err.Error())
+			}
+		}
+
+		// Close channels
+		close(watchChan)
+		close(replyChan)
+
+		// Close watcher
+		err := watcher.Close()
+		if err != nil {
+			fs.Errorf(f, "Failed to close watcher: %s", err)
+		}
+	}()
+
+	// Start goroutine to establish watchers
+	go func() {
+		for {
+			path, ok := <-watchChan
+			if !ok {
+				break
+			}
+
+			// Is this the initial watch?
+			initial := path == f.root
+
+			// Determine entry path
+			entryPath := ""
+			if !initial {
+				entryPath, err = filepath.Rel(f.root, path)
+				if err != nil {
+					// Not in this remote
+					replyChan <- true
+					continue
+				}
+			}
+
+			// Determine entry type
+			entryType := fs.EntryObject
+			if initial {
+				// Known to be a directory, but also cannot Lstat() some mounts
+				entryType = fs.EntryDirectory
+			} else {
+				info, err := os.Lstat(path)
+				if err != nil {
+					fs.Errorf(f, "Failed to stat %s, already removed? %s", path, err)
+					replyChan <- true
+					continue
+				} else if info.IsDir() {
+					entryType = fs.EntryDirectory
+				}
+				changed[entryPath] = time.Now()
+			}
+
+			if entryType == fs.EntryDirectory {
+				// Recursively watch the directory
+				err := watcher.Add(path)
+				if err != nil {
+					fs.Errorf(f, "Failed to start watching %s, already removed? %s", path, err)
+				} else {
+					fs.Logf(f, "Started watching %s", path)
+				}
+				err = walk.Walk(ctx, f, entryPath, false, -1, func(entryPath string, entries fs.DirEntries, err error) error {
+					if err != nil {
+						// The entry has already been removed, and we do not know what
+						// type it was. It can be ignored, as this means it has been both
+						// created and removed since the last tick, which will not change
+						// the diff at the next tick.
+						fs.Errorf(f, "Failed to walk %s, already removed? %s", path, err)
+					}
+					for _, d := range entries {
+						entryPath := d.Remote()
+						path := filepath.Join(f.root, entryPath)
+						info, err := os.Lstat(path)
+						if err != nil {
+							fs.Errorf(f, "Failed to stat %s, already removed? %s", path, err)
+							continue
+						}
+						if !initial {
+							changed[entryPath] = time.Now()
+						}
+						if info.IsDir() {
+							// Watch the directory.
+							//
+							// Establishing a watch on a directory before listing its
+							// contents ensures that no entries are missed and all changes
+							// are notified, even for entries created or modified while
+							// the watch is being established.
+							//
+							// An entry may be created between establishing the watch on
+							// the directory and listing the directory. In this case it is
+							// marked as changed both by this walk and the subsequent
+							// handling of the associated filesystem event. Because
+							// changes are accumulated up to the next tick, however, only
+							// a single notification is sent at the next tick.
+							//
+							// If an entry exists when the walk begins, but is removed
+							// before the walk reaches it, it is as though that entry
+							// never existed. But as both occur since the last tick, this
+							// does not affect the diff at the next tick.
+							err := watcher.Add(path)
+							if err != nil {
+								fs.Errorf(f, "Failed to start watching %s, already removed? %s", entryPath, err)
+							} else {
+								fs.Logf(f, "Started watching %s", entryPath)
+							}
+						}
+					}
+					return nil
+				})
+				if err != nil {
+					fs.Errorf(f, "Failed to walk %s, already removed? %s", entryPath, err)
+				}
+			}
+			replyChan <- true
+		}
+	}()
+
+	// Recursively watch all subdirectories from the root
+	watchChan <- f.root
+
+	// Wait until initial watch is established before returning
+	<-replyChan
+}

backend/local/changenotify_windows.go

@@ -0,0 +1,150 @@
+//go:build windows
+
+package local
+
+import (
+	"context"
+	"path/filepath"
+	"time"
+	_ "unsafe" // use go:linkname
+
+	"github.com/fsnotify/fsnotify"
+	"github.com/rclone/rclone/fs"
+	"github.com/rclone/rclone/fs/walk"
+)
+
+// Hack to enable recursive watchers in fsnotify, which are available for
+// Windows and Linux (although not quite what is needed for Linux here), but
+// not yet Mac, hence there is no public interface for this in fsnotify just
+// yet. This is currently only needed for, and enabled for, Windows builds.
+//
+// Setting fsnotify.enableRecurse to true enables recursive handling: paths
+// that end with with \... or /... as watched recursively.
+
+//go:linkname enableRecurse github.com/fsnotify/fsnotify.enableRecurse
+var enableRecurse bool
+
+// ChangeNotify calls the passed function with a path that has had changes.
+// Close the returned channel to stop being notified.
+func (f *Fs) ChangeNotify(ctx context.Context, notifyFunc func(string, fs.EntryType), pollIntervalChan <-chan time.Duration) {
+
+	// Create new watcher
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		fs.Errorf(f, "Failed to create watcher: %s", err)
+		return
+	}
+
+	// Recursive watch of base directory. This is indicated by appending \...
+	// or /... to the path.
+	enableRecurse = true
+	err = watcher.Add(filepath.Join(f.root, "..."))
+	if err != nil {
+		fs.Errorf(f, "Failed to start watching %s: %s", f.root, err)
+	} else {
+		fs.Debugf(f, "Started watching %s", f.root)
+	}
+
+	// Files and directories changed in the last poll window, mapped to the
+	// time at which notification of the change was received.
+	changed := make(map[string]time.Time)
+
+	// Start goroutine to handle filesystem events
+	go func() {
+		// Polling is imitated by accumulating events between ticks. While
+		// notifyFunc() could be called immediately on each filesystem event,
+		// accumulating turns out to have some advantages in accurately keeping
+		// track of entry types (i.e. file or directory), under the
+		// interpretation that the notifications sent at each tick are a diff of
+		// the state of the filesystem at that tick compared to the previous. It
+		// is also assumed by some tests.
+		var ticker *time.Ticker
+		var tickerC <-chan time.Time
+
+	loop:
+		for {
+			select {
+			case pollInterval, ok := <-pollIntervalChan:
+				// Update ticker
+				if !ok {
+					if ticker != nil {
+						ticker.Stop()
+					}
+					break loop
+				}
+				if ticker != nil {
+					ticker.Stop()
+					ticker, tickerC = nil, nil
+				}
+				if pollInterval != 0 {
+					ticker = time.NewTicker(pollInterval)
+					tickerC = ticker.C
+				}
+			case <-tickerC:
+				// Notify for all paths that have changed since the last sync, and
+				// which were changed at least 1/10 of a second (1e8 nanoseconds)
+				// ago. The lag is for de-duping purposes during long writes, which
+				// can consist of multiple write notifications in quick succession.
+				cutoff := time.Now().Add(-1e8)
+				for entryPath, entryTime := range changed {
+					if entryTime.Before(cutoff) {
+						notifyFunc(filepath.ToSlash(entryPath), fs.EntryUncertain)
+						delete(changed, entryPath)
+					}
+				}
+			case event, ok := <-watcher.Events:
+				if !ok {
+					break loop
+				}
+				if event.Has(fsnotify.Create) {
+					fs.Debugf(f, "Create: %s", event.Name)
+				}
+				if event.Has(fsnotify.Remove) {
+					fs.Debugf(f, "Remove: %s", event.Name)
+				}
+				if event.Has(fsnotify.Rename) {
+					fs.Debugf(f, "Rename: %s", event.Name)
+				}
+				if event.Has(fsnotify.Write) {
+					fs.Debugf(f, "Write: %s", event.Name)
+				}
+				if event.Has(fsnotify.Chmod) {
+					fs.Debugf(f, "Chmod: %s", event.Name)
+				}
+				entryPath, _ := filepath.Rel(f.root, event.Name)
+				changed[entryPath] = time.Now()
+
+				if event.Has(fsnotify.Create) {
+					err = walk.Walk(ctx, f, entryPath, false, -1, func(entryPath string, entries fs.DirEntries, err error) error {
+						if err != nil {
+							// The entry has already been removed, and we do not know what
+							// type it was. It can be ignored, as this means it has been both
+							// created and removed since the last tick, which will not change
+							// the diff at the next tick.
+							fs.Errorf(f, "Failed to walk %s, already removed? %s", entryPath, err)
+						}
+						for _, d := range entries {
+							entryPath := d.Remote()
+							changed[entryPath] = time.Now()
+						}
+						return nil
+					})
+					if err != nil {
+						fs.Errorf(f, "Failed to walk %s, already removed? %s", entryPath, err)
+					}
+				}
+			case err, ok := <-watcher.Errors:
+				if !ok {
+					break loop
+				}
+				fs.Errorf(f, "Error: %s", err.Error())
+			}
+		}
+
+		// Close watcher
+		err := watcher.Close()
+		if err != nil {
+			fs.Errorf(f, "Failed to close watcher: %s", err)
+		}
+	}()
+}

backend/local/local.go

@@ -1700,6 +1700,7 @@ var (
 	_ fs.OpenWriterAter  = &Fs{}
 	_ fs.DirSetModTimer  = &Fs{}
 	_ fs.MkdirMetadataer = &Fs{}
+	_ fs.ChangeNotifier  = &Fs{}
 	_ fs.Object          = &Object{}
 	_ fs.Metadataer      = &Object{}
 	_ fs.SetMetadataer   = &Object{}

fs/features.go

@@ -589,6 +589,9 @@ const (
 	EntryDirectory EntryType = iota // 0
 	// EntryObject should be used to classify remote paths in objects
 	EntryObject // 1
+	// EntryUncertain should be used when a remote path cannot reliably or
+	// efficiently be classified as EntryDirectory or EntryObject
+	EntryUncertain // 2
 )
 
 // UnWrapper is an optional interfaces for Fs

fstest/fstests/fstests.go

@@ -21,6 +21,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -874,17 +875,23 @@ func Run(t *testing.T, opt *Opt) {
 			pollInterval := make(chan time.Duration)
 			dirChanges := map[string]struct{}{}
 			objChanges := map[string]struct{}{}
+			uncChanges := map[string]struct{}{}
+			var mutex sync.Mutex
 			doChangeNotify(ctx, func(x string, e fs.EntryType) {
 				fs.Debugf(nil, "doChangeNotify(%q, %+v)", x, e)
 				if strings.HasPrefix(x, file1.Path[:5]) || strings.HasPrefix(x, file2.Path[:5]) {
 					fs.Debugf(nil, "Ignoring notify for file1 or file2: %q, %v", x, e)
 					return
 				}
+				mutex.Lock()
 				if e == fs.EntryDirectory {
 					dirChanges[x] = struct{}{}
 				} else if e == fs.EntryObject {
 					objChanges[x] = struct{}{}
+				} else if e == fs.EntryUncertain {
+					uncChanges[x] = struct{}{}
 				}
+				mutex.Unlock()
 			}, pollInterval)
 			defer func() { close(pollInterval) }()
 			pollInterval <- time.Second
@@ -922,17 +929,20 @@ func Run(t *testing.T, opt *Opt) {
 			// Wait a little while for the changes to come in
 			wantDirChanges := []string{"dir/subdir1", "dir/subdir3", "dir/subdir2"}
 			wantObjChanges := []string{"dir/file2", "dir/file4", "dir/file3"}
+			wantUncChanges := append(wantDirChanges, wantObjChanges...)
 			ok := false
 			for tries := 1; tries < 10; tries++ {
-				ok = contains(dirChanges, wantDirChanges) && contains(objChanges, wantObjChanges)
+				mutex.Lock()
+				ok = (contains(dirChanges, wantDirChanges) && contains(objChanges, wantObjChanges)) || contains(uncChanges, wantUncChanges)
+				mutex.Unlock()
 				if ok {
 					break
 				}
-				t.Logf("Try %d/10 waiting for dirChanges and objChanges", tries)
+				t.Logf("Try %d/10 waiting for dirChanges, objChanges, and uncChanges", tries)
 				time.Sleep(3 * time.Second)
 			}
 			if !ok {
-				t.Errorf("%+v does not contain %+v or \n%+v does not contain %+v", dirChanges, wantDirChanges, objChanges, wantObjChanges)
+				t.Errorf("%+v does not contain %+v or \n%+v does not contain %+v, and %+v does not contain %+v", dirChanges, wantDirChanges, objChanges, wantObjChanges, uncChanges, wantUncChanges)
 			}
 
 			// tidy up afterwards

go.mod

@@ -144,6 +144,7 @@ require (
 	github.com/emersion/go-vcard v0.0.0-20230815062825-8fda7d206ec9 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/flynn/noise v1.0.1 // indirect
+	github.com/fsnotify/fsnotify v1.8.0 // indirect
 	github.com/gdamore/encoding v1.0.1 // indirect
 	github.com/geoffgarside/ber v1.1.0 // indirect
 	github.com/go-ini/ini v1.67.0 // indirect

go.sum

@@ -241,6 +241,8 @@ github.com/frankban/quicktest v1.14.4 h1:g2rn0vABPOOXmZUj+vbmUp0lPoXEMuhTpIluN0X
 github.com/frankban/quicktest v1.14.4/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
 github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=
 github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM=
+github.com/fsnotify/fsnotify v1.8.0 h1:dAwr6QBTBZIkG8roQaJjGof0pp0EeF+tNV7YBP3F/8M=
+github.com/fsnotify/fsnotify v1.8.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
 github.com/gabriel-vasile/mimetype v1.4.8 h1:FfZ3gj38NjllZIeJAmMhr+qKL8Wu+nOoI3GqacKw1NM=
 github.com/gabriel-vasile/mimetype v1.4.8/go.mod h1:ByKUIKGjh1ODkGM1asKUbQZOLGrPjydw3hYPU2YU9t8=
 github.com/gdamore/encoding v1.0.1 h1:YzKZckdBL6jVt2Gc+5p82qhrGiqMdG/eNs6Wy0u3Uhw=

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"
@@ -97,7 +98,15 @@ func TestRcPollInterval(t *testing.T) {
 	}
 	out, err := call.Fn(context.Background(), nil)
 	require.NoError(t, err)
-	assert.Equal(t, rc.Params{}, out)
+	assert.Equal(t, rc.Params{
+		"enabled": true,
+		"interval": map[string]interface{}{
+			"raw":     fs.Duration(60000000000),
+			"seconds": time.Duration(60),
+			"string":  "1m0s",
+		},
+		"supported": true,
+	}, out)
 	// FIXME needs more tests
 }