vfs: rename files in cache and cancel uploads on directory rename

Before this change rclone did not cancel an uploads or rename the
cached files in the directory cache when a directory was renamed.

This caused issues with uploads arriving in the wrong place on bucket
based file systems.

See: https://forum.rclone.org/t/after-a-directory-renmane-using-mv-files-are-not-visible-any-longer/22797
This commit is contained in:
Nick Craig-Wood 2021-03-17 09:48:25 +00:00
parent 4cc2a7f342
commit a4c4ddf052
3 changed files with 62 additions and 10 deletions

View File

@ -322,11 +322,20 @@ func (d *Dir) rename(newParent *Dir, fsDir fs.Directory) {
d.modTime = fsDir.ModTime(context.TODO())
d.modTimeMu.Unlock()
d.mu.Lock()
oldPath := d.path
d.parent = newParent
d.entry = fsDir
d.path = fsDir.Remote()
newPath := d.path
d.read = time.Time{}
d.mu.Unlock()
// Rename in the cache
if d.vfs.cache != nil && d.vfs.cache.DirExists(oldPath) {
if err := d.vfs.cache.DirRename(oldPath, newPath); err != nil {
fs.Infof(d, "Dir.Rename failed in Cache: %v", err)
}
}
}
// addObject adds a new object or directory to the directory

View File

@ -135,7 +135,7 @@ func New(ctx context.Context, fremote fs.Fs, opt *vfscommon.Options, avFn AddVir
}
// Remove any empty directories
c.purgeEmptyDirs()
c.purgeEmptyDirs("", true)
// Create a channel for cleaner to be kicked upon out of space con
c.kick = make(chan struct{}, 1)
@ -344,6 +344,49 @@ func (c *Cache) Rename(name string, newName string, newObj fs.Object) (err error
return nil
}
// DirExists checks to see if the directory exists in the cache or not.
func (c *Cache) DirExists(name string) bool {
path := c.toOSPath(name)
_, err := os.Stat(path)
return err == nil
}
// DirRename the dir in cache
func (c *Cache) DirRename(oldDirName string, newDirName string) (err error) {
// Make sure names are / suffixed for reading keys out of c.item
if !strings.HasSuffix(oldDirName, "/") {
oldDirName += "/"
}
if !strings.HasSuffix(newDirName, "/") {
newDirName += "/"
}
// Find all items to rename
var renames []string
c.mu.Lock()
for itemName := range c.item {
if strings.HasPrefix(itemName, oldDirName) {
renames = append(renames, itemName)
}
}
c.mu.Unlock()
// Rename the items
for _, itemName := range renames {
newPath := newDirName + itemName[len(oldDirName):]
renameErr := c.Rename(itemName, newPath, nil)
if renameErr != nil {
err = renameErr
}
}
// Old path should be empty now so remove it
c.purgeEmptyDirs(oldDirName[:len(oldDirName)-1], false)
fs.Infof(oldDirName, "vfs cache: renamed dir in cache to %q", newDirName)
return err
}
// Remove should be called if name is deleted
//
// This returns true if the file was in the transfer queue so may not
@ -555,15 +598,15 @@ func (c *Cache) purgeOld(maxAge time.Duration) {
}
// Purge any empty directories
func (c *Cache) purgeEmptyDirs() {
func (c *Cache) purgeEmptyDirs(dir string, leaveRoot bool) {
ctx := context.Background()
err := operations.Rmdirs(ctx, c.fcache, "", true)
err := operations.Rmdirs(ctx, c.fcache, dir, leaveRoot)
if err != nil {
fs.Errorf(c.fcache, "vfs cache: failed to remove empty directories from cache: %v", err)
fs.Errorf(c.fcache, "vfs cache: failed to remove empty directories from cache path %q: %v", dir, err)
}
err = operations.Rmdirs(ctx, c.fcacheMeta, "", true)
err = operations.Rmdirs(ctx, c.fcacheMeta, dir, leaveRoot)
if err != nil {
fs.Errorf(c.fcache, "vfs cache: failed to remove empty directories from metadata cache: %v", err)
fs.Errorf(c.fcache, "vfs cache: failed to remove empty directories from metadata cache path %q: %v", dir, err)
}
}

View File

@ -279,7 +279,7 @@ func TestCacheOpenMkdir(t *testing.T) {
// clean the cache
c.purgeOld(-10 * time.Second)
c.purgeEmptyDirs()
c.purgeEmptyDirs("", true)
assert.Equal(t, []string(nil), itemAsString(c))
@ -407,7 +407,7 @@ func TestCachePurgeOverQuota(t *testing.T) {
// Check only potato2 removed to get below quota
c.purgeOverQuota(10)
assert.Equal(t, int64(5), c.used)
c.purgeEmptyDirs()
c.purgeEmptyDirs("", true)
assert.Equal(t, []string{
`name="sub/dir/potato" opens=0 size=5`,
@ -416,7 +416,7 @@ func TestCachePurgeOverQuota(t *testing.T) {
// Now purge everything
c.purgeOverQuota(1)
assert.Equal(t, int64(0), c.used)
c.purgeEmptyDirs()
c.purgeEmptyDirs("", true)
assert.Equal(t, []string(nil), itemAsString(c))
@ -485,7 +485,7 @@ func TestCachePurgeClean(t *testing.T) {
// So we use purgeOverQuota here for the cleanup.
c.purgeOverQuota(1)
c.purgeEmptyDirs()
c.purgeEmptyDirs("", true)
assert.Equal(t, []string(nil), itemAsString(c))
}