Clean IDCache properly

changelog/unreleased/clean-decomposedfs-idcache.md

@@ -0,0 +1,7 @@
+Bugfix: Clean IDCache properly
+
+decomposedfs' subpackage `tree` uses an idCache to avoid reading too often from disc. In case of a `move` or `delete` this cache was
+properly cleaned, but when renaming a file (= move with same parent) the cache wasn't cleaned. This lead to strange behaviour when
+uploading files with the same name and renaming them
+
+https://github.com/cs3org/reva/pull/3903

pkg/storage/utils/decomposedfs/decomposedfs.go

@@ -56,9 +56,11 @@ import (
 	"github.com/cs3org/reva/v2/pkg/storage/utils/filelocks"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/templates"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
+	"github.com/cs3org/reva/v2/pkg/store"
 	"github.com/cs3org/reva/v2/pkg/utils"
 	"github.com/jellydator/ttlcache/v2"
 	"github.com/pkg/errors"
+	microstore "go-micro.dev/v4/store"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -117,8 +119,14 @@ func NewDefault(m map[string]interface{}, bs tree.Blobstore, es events.Stream) (
 		return nil, fmt.Errorf("unknown metadata backend %s, only 'messagepack' or 'xattrs' (default) supported", o.MetadataBackend)
 	}
 
-	tp := tree.New(lu, bs, o)
-
+	tp := tree.New(lu, bs, o, store.Create(
+		store.Store(o.IDCache.Store),
+		store.TTL(time.Duration(o.IDCache.TTL)*time.Second),
+		store.Size(o.IDCache.Size),
+		microstore.Nodes(o.IDCache.Nodes...),
+		microstore.Database(o.IDCache.Database),
+		microstore.Table(o.IDCache.Table),
+	))
 	permissionsClient, err := pool.GetPermissionsClient(o.PermissionsSVC, pool.WithTLSMode(o.PermTLSMode))
 	if err != nil {
 		return nil, err

pkg/storage/utils/decomposedfs/options/options.go

@@ -69,6 +69,7 @@ type Options struct {
 
 	StatCache         cache.Config `mapstructure:"statcache"`
 	FileMetadataCache cache.Config `mapstructure:"filemetadatacache"`
+	IDCache           cache.Config `mapstructure:"idcache"`
 
 	MaxAcquireLockCycles    int `mapstructure:"max_acquire_lock_cycles"`
 	LockCycleDurationFactor int `mapstructure:"lock_cycle_duration_factor"`

pkg/storage/utils/decomposedfs/testhelpers/helpers.go

@@ -28,6 +28,7 @@ import (
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/metadata/prefixes"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
+	"github.com/cs3org/reva/v2/pkg/store"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/mock"
 
@@ -156,7 +157,7 @@ func NewTestEnv(config map[string]interface{}) (*TestEnv, error) {
 	permissions := &mocks.PermissionsChecker{}
 	cs3permissionsclient := &mocks.CS3PermissionsClient{}
 	bs := &treemocks.Blobstore{}
-	tree := tree.New(lu, bs, o)
+	tree := tree.New(lu, bs, o, store.Create())
 	fs, err := decomposedfs.New(o, lu, decomposedfs.NewPermissions(permissions, cs3permissionsclient), tree, nil)
 	if err != nil {
 		return nil, err

pkg/storage/utils/decomposedfs/tree/tree.go

@@ -32,7 +32,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/bluele/gcache"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/cs3org/reva/v2/pkg/appctx"
 	"github.com/cs3org/reva/v2/pkg/errtypes"
@@ -47,6 +46,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/rogpeppe/go-internal/lockedfile"
 	"github.com/rs/zerolog/log"
+	"go-micro.dev/v4/store"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -80,19 +80,19 @@ type Tree struct {
 
 	options *options.Options
 
-	idCache gcache.Cache
+	idCache store.Store
 }
 
 // PermissionCheckFunc defined a function used to check resource permissions
 type PermissionCheckFunc func(rp *provider.ResourcePermissions) bool
 
 // New returns a new instance of Tree
-func New(lu PathLookup, bs Blobstore, o *options.Options) *Tree {
+func New(lu PathLookup, bs Blobstore, o *options.Options, cache store.Store) *Tree {
 	return &Tree{
 		lookup:    lu,
 		blobstore: bs,
 		options:   o,
-		idCache:   gcache.New(1_000_000).LRU().Build(),
+		idCache:   cache,
 	}
 }
 
@@ -234,6 +234,9 @@ func (t *Tree) Move(ctx context.Context, oldNode *node.Node, newNode *node.Node)
 		}
 	}
 
+	// remove cache entry in any case to avoid inconsistencies
+	_ = t.idCache.Delete(filepath.Join(oldNode.ParentPath(), oldNode.Name))
+
 	// Always target the old node ID for xattr updates.
 	// The new node id is empty if the target does not exist
 	// and we need to overwrite the new one when overwriting an existing path.
@@ -271,7 +274,6 @@ func (t *Tree) Move(ctx context.Context, oldNode *node.Node, newNode *node.Node)
 	if err != nil {
 		return errors.Wrap(err, "Decomposedfs: could not move child")
 	}
-	t.idCache.Remove(filepath.Join(oldNode.ParentPath(), oldNode.Name))
 
 	// update target parentid and name
 	attribs := node.Attributes{}
@@ -361,16 +363,14 @@ func (t *Tree) ListFolder(ctx context.Context, n *node.Node) ([]*node.Node, erro
 	for i := 0; i < numWorkers; i++ {
 		g.Go(func() error {
 			for name := range work {
-				nodeID := ""
 				path := filepath.Join(dir, name)
-				if val, err := t.idCache.Get(path); err == nil {
-					nodeID = val.(string)
-				} else {
+				nodeID := getNodeIDFromCache(path, t.idCache)
+				if nodeID == "" {
 					nodeID, err = readChildNodeFromLink(path)
 					if err != nil {
 						return err
 					}
-					err = t.idCache.Set(path, nodeID)
+					err = storeNodeIDInCache(path, nodeID, t.idCache)
 					if err != nil {
 						return err
 					}
@@ -416,6 +416,10 @@ func (t *Tree) ListFolder(ctx context.Context, n *node.Node) ([]*node.Node, erro
 
 // Delete deletes a node in the tree by moving it to the trash
 func (t *Tree) Delete(ctx context.Context, n *node.Node) (err error) {
+	path := filepath.Join(n.ParentPath(), n.Name)
+	// remove entry from cache immediately to avoid inconsistencies
+	_ = t.idCache.Delete(path)
+
 	deletingSharedResource := ctx.Value(appctx.DeletingSharedResource)
 
 	if deletingSharedResource != nil && deletingSharedResource.(bool) {
@@ -492,10 +496,7 @@ func (t *Tree) Delete(ctx context.Context, n *node.Node) (err error) {
 	_ = os.Remove(n.LockFilePath())
 
 	// finally remove the entry from the parent dir
-	path := filepath.Join(n.ParentPath(), n.Name)
-	err = os.Remove(path)
-	t.idCache.Remove(path)
-	if err != nil {
+	if err = os.Remove(path); err != nil {
 		// To roll back changes
 		// TODO revert the rename
 		// TODO remove symlink
@@ -980,3 +981,18 @@ func (t *Tree) readRecycleItem(ctx context.Context, spaceID, key, path string) (
 
 	return
 }
+
+func getNodeIDFromCache(path string, cache store.Store) string {
+	recs, err := cache.Read(path)
+	if err == nil && len(recs) > 0 {
+		return string(recs[0].Value)
+	}
+	return ""
+}
+
+func storeNodeIDInCache(path string, nodeID string, cache store.Store) error {
+	return cache.Write(&store.Record{
+		Key:   path,
+		Value: []byte(nodeID),
+	})
+}

pkg/storage/utils/decomposedfs/upload_async_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/tree"
 	treemocks "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/tree/mocks"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
+	"github.com/cs3org/reva/v2/pkg/store"
 	"github.com/cs3org/reva/v2/pkg/utils"
 	"github.com/cs3org/reva/v2/tests/helpers"
 	"github.com/stretchr/testify/mock"
@@ -105,7 +106,7 @@ var _ = Describe("Async file uploads", Ordered, func() {
 
 		// setup fs
 		pub, con = make(chan interface{}), make(chan interface{})
-		tree := tree.New(lu, bs, o)
+		tree := tree.New(lu, bs, o, store.Create())
 		fs, err = New(o, lu, NewPermissions(permissions, cs3permissionsclient), tree, stream.Chan{pub, con})
 		Expect(err).ToNot(HaveOccurred())
 

pkg/storage/utils/decomposedfs/upload_test.go

@@ -41,6 +41,7 @@ import (
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/tree"
 	treemocks "github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/tree/mocks"
 	"github.com/cs3org/reva/v2/pkg/storagespace"
+	"github.com/cs3org/reva/v2/pkg/store"
 	"github.com/cs3org/reva/v2/tests/helpers"
 	"github.com/stretchr/testify/mock"
 
@@ -119,7 +120,7 @@ var _ = Describe("File uploads", func() {
 			AddGrant: true,
 		}, nil).Times(1)
 		var err error
-		tree := tree.New(lu, bs, o)
+		tree := tree.New(lu, bs, o, store.Create())
 		fs, err = decomposedfs.New(o, lu, decomposedfs.NewPermissions(permissions, cs3permissionsclient), tree, nil)
 		Expect(err).ToNot(HaveOccurred())