Cleanup and improve the caching config (#6148)

* Cleanup and improve the caching config * bump reva Signed-off-by: Jörn Friedrich Dreyer <jfd@butonic.de> * disable stat cache Signed-off-by: Jörn Friedrich Dreyer <jfd@butonic.de> * Bump reva * Linter fixes --------- Signed-off-by: Jörn Friedrich Dreyer <jfd@butonic.de> Co-authored-by: Jörn Friedrich Dreyer <jfd@butonic.de>
2023-04-27 12:35:30 +02:00
parent d563b63d8f
commit 129489203b
29 changed files with 277 additions and 253 deletions
@@ -68,15 +68,21 @@ type config struct {
 	DataTransfersFolder          string                            `mapstructure:"data_transfers_folder"`
 	TokenManagers                map[string]map[string]interface{} `mapstructure:"token_managers"`
 	AllowedUserAgents            map[string][]string               `mapstructure:"allowed_user_agents"` // map[path][]user-agent
-	CacheStore                   string                            `mapstructure:"cache_store"`
-	CacheNodes                   []string                          `mapstructure:"cache_nodes"`
-	CacheDatabase                string                            `mapstructure:"cache_database"`
-	CreateHomeCacheTTL           int                               `mapstructure:"create_home_cache_ttl"`
-	CreateHomeCacheSize          int                               `mapstructure:"create_home_cache_size"`
-	ProviderCacheTTL             int                               `mapstructure:"provider_cache_ttl"`
-	ProviderCacheSize            int                               `mapstructure:"provider_cache_size"`
+	StatCacheStore               string                            `mapstructure:"stat_cache_store"`
+	StatCacheNodes               []string                          `mapstructure:"stat_cache_nodes"`
+	StatCacheDatabase            string                            `mapstructure:"stat_cache_database"`
 	StatCacheTTL                 int                               `mapstructure:"stat_cache_ttl"`
 	StatCacheSize                int                               `mapstructure:"stat_cache_size"`
+	CreateHomeCacheStore         string                            `mapstructure:"create_home_cache_store"`
+	CreateHomeCacheNodes         []string                          `mapstructure:"create_home_cache_nodes"`
+	CreateHomeCacheDatabase      string                            `mapstructure:"create_home_cache_database"`
+	CreateHomeCacheTTL           int                               `mapstructure:"create_home_cache_ttl"`
+	CreateHomeCacheSize          int                               `mapstructure:"create_home_cache_size"`
+	ProviderCacheStore           string                            `mapstructure:"provider_cache_store"`
+	ProviderCacheNodes           []string                          `mapstructure:"provider_cache_nodes"`
+	ProviderCacheDatabase        string                            `mapstructure:"provider_cache_database"`
+	ProviderCacheTTL             int                               `mapstructure:"provider_cache_ttl"`
+	ProviderCacheSize            int                               `mapstructure:"provider_cache_size"`
 	UseCommonSpaceRootShareLogic bool                              `mapstructure:"use_common_space_root_share_logic"`
 }

@@ -124,12 +130,28 @@ func (c *config) init() {
 	}

 	// caching needs to be explicitly enabled
-	if c.CacheStore == "" {
-		c.CacheStore = "noop"
+	if c.StatCacheStore == "" {
+		c.StatCacheStore = "noop"
 	}

-	if c.CacheDatabase == "" {
-		c.CacheDatabase = "reva"
+	if c.StatCacheDatabase == "" {
+		c.StatCacheDatabase = "reva"
+	}
+
+	if c.ProviderCacheStore == "" {
+		c.ProviderCacheStore = "noop"
+	}
+
+	if c.ProviderCacheDatabase == "" {
+		c.ProviderCacheDatabase = "reva"
+	}
+
+	if c.CreateHomeCacheStore == "" {
+		c.CreateHomeCacheStore = "noop"
+	}
+
+	if c.CreateHomeCacheDatabase == "" {
+		c.CreateHomeCacheDatabase = "reva"
 	}
 }

@@ -169,10 +191,10 @@ func New(m map[string]interface{}, ss *grpc.Server) (rgrpc.Service, error) {
 		c:                        c,
 		dataGatewayURL:           *u,
 		tokenmgr:                 tokenManager,
-		statCache:                cache.GetStatCache(c.CacheStore, c.CacheNodes, c.CacheDatabase, "stat", time.Duration(c.StatCacheTTL)*time.Second, c.StatCacheSize),
-		providerCache:            cache.GetProviderCache(c.CacheStore, c.CacheNodes, c.CacheDatabase, "provider", time.Duration(c.ProviderCacheTTL)*time.Second, c.ProviderCacheSize),
-		createHomeCache:          cache.GetCreateHomeCache(c.CacheStore, c.CacheNodes, c.CacheDatabase, "createHome", time.Duration(c.CreateHomeCacheTTL)*time.Second, c.CreateHomeCacheSize),
-		createPersonalSpaceCache: cache.GetCreatePersonalSpaceCache(c.CacheStore, c.CacheNodes, c.CacheDatabase, "createPersonalSpace", time.Duration(c.CreateHomeCacheTTL)*time.Second, c.CreateHomeCacheSize),
+		statCache:                cache.GetStatCache(c.StatCacheStore, c.StatCacheNodes, c.StatCacheDatabase, "stat", time.Duration(c.StatCacheTTL)*time.Second, c.StatCacheSize),
+		providerCache:            cache.GetProviderCache(c.ProviderCacheStore, c.ProviderCacheNodes, c.ProviderCacheDatabase, "provider", time.Duration(c.ProviderCacheTTL)*time.Second, c.ProviderCacheSize),
+		createHomeCache:          cache.GetCreateHomeCache(c.CreateHomeCacheStore, c.CreateHomeCacheNodes, c.CreateHomeCacheDatabase, "createHome", time.Duration(c.CreateHomeCacheTTL)*time.Second, c.CreateHomeCacheSize),
+		createPersonalSpaceCache: cache.GetCreatePersonalSpaceCache(c.CreateHomeCacheStore, c.CreateHomeCacheNodes, c.CreateHomeCacheDatabase, "createPersonalSpace", time.Duration(c.CreateHomeCacheTTL)*time.Second, c.CreateHomeCacheSize),
 	}

 	return s, nil
@@ -37,12 +37,12 @@ type Config struct {
 	AdditionalInfoAttribute               string                            `mapstructure:"additional_info_attribute"`
 	CacheWarmupDriver                     string                            `mapstructure:"cache_warmup_driver"`
 	CacheWarmupDrivers                    map[string]map[string]interface{} `mapstructure:"cache_warmup_drivers"`
-	ResourceInfoCacheStore                string                            `mapstructure:"resource_info_cache_store"`
-	ResourceInfoCacheNodes                []string                          `mapstructure:"resource_info_cache_nodes"`
-	ResourceInfoCacheDatabase             string                            `mapstructure:"resource_info_cache_database"`
-	ResourceInfoCacheTable                string                            `mapstructure:"resource_info_cache_table"`
-	ResourceInfoCacheTTL                  int                               `mapstructure:"resource_info_cache_ttl"`
-	ResourceInfoCacheSize                 int                               `mapstructure:"resource_info_cache_size"`
+	StatCacheStore                        string                            `mapstructure:"stat_cache_store"`
+	StatCacheNodes                        []string                          `mapstructure:"stat_cache_nodes"`
+	StatCacheDatabase                     string                            `mapstructure:"stat_cache_database"`
+	StatCacheTable                        string                            `mapstructure:"stat_cache_table"`
+	StatCacheTTL                          int                               `mapstructure:"stat_cache_ttl"`
+	StatCacheSize                         int                               `mapstructure:"stat_cache_size"`
 	UserIdentifierCacheTTL                int                               `mapstructure:"user_identifier_cache_ttl"`
 	MachineAuthAPIKey                     string                            `mapstructure:"machine_auth_apikey"`
 	SkipUpdatingExistingSharesMountpoints bool                              `mapstructure:"skip_updating_existing_shares_mountpoint"`
@@ -130,7 +130,7 @@ func (h *Handler) Init(c *config.Config) {
 	h.deniable = c.EnableDenials
 	h.resharing = resharing(c)

-	h.statCache = cache.GetStatCache(c.ResourceInfoCacheStore, c.ResourceInfoCacheNodes, c.ResourceInfoCacheDatabase, "stat", time.Duration(c.ResourceInfoCacheTTL)*time.Second, c.ResourceInfoCacheSize)
+	h.statCache = cache.GetStatCache(c.StatCacheStore, c.StatCacheNodes, c.StatCacheDatabase, "stat", time.Duration(c.StatCacheTTL)*time.Second, c.StatCacheSize)
 	if c.CacheWarmupDriver != "" {
 		cwm, err := getCacheWarmupManager(c)
 		if err == nil {
@@ -67,9 +67,9 @@ func New(m map[string]interface{}, log *zerolog.Logger) (global.Service, error)
 		return nil, err
 	}

-	if conf.CacheWarmupDriver == "first-request" && conf.ResourceInfoCacheTTL > 0 {
+	if conf.CacheWarmupDriver == "first-request" && conf.StatCacheStore != "noop" {
 		s.warmupCacheTracker = ttlcache.NewCache()
-		_ = s.warmupCacheTracker.SetTTL(time.Second * time.Duration(conf.ResourceInfoCacheTTL))
+		_ = s.warmupCacheTracker.SetTTL(time.Second * time.Duration(conf.StatCacheTTL))
 	}

 	return s, nil
@@ -19,7 +19,6 @@
 package cache

 import (
-	"strings"
 	"time"
 )

@@ -40,15 +39,5 @@ func NewFileMetadataCache(store string, nodes []string, database, table string,

 // RemoveMetadata removes a reference from the metadata cache
 func (c *fileMetadataCache) RemoveMetadata(path string) error {
-	keys, err := c.List()
-	if err != nil {
-		return err
-	}
-
-	for _, key := range keys {
-		if strings.HasPrefix(key, path) {
-			_ = c.Delete(key)
-		}
-	}
-	return nil
+	return c.s.Delete(path)
 }
@@ -19,11 +19,12 @@
 package cache

 import (
-	"strings"
+	"sync"
 	"time"

 	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	"go-micro.dev/v4/store"
 )

 // ProviderCache can invalidate all provider related cache entries
@@ -47,20 +48,22 @@ func (c providerCache) RemoveListStorageProviders(res *provider.ResourceId) {
 	if res == nil {
 		return
 	}
-	sid := res.SpaceId

-	keys, err := c.List()
+	keys, err := c.List(store.ListSuffix(res.SpaceId), store.ListLimit(100))
 	if err != nil {
 		// FIXME log error
 		return
 	}
-	// FIXME add context option to List, Read and Write to upstream
+
+	wg := sync.WaitGroup{}
 	for _, key := range keys {
-		if strings.Contains(key, sid) {
-			_ = c.Delete(key)
-			continue
-		}
+		wg.Add(1)
+		go func(k string) {
+			defer wg.Done()
+			_ = c.Delete(k)
+		}(key)
 	}
+	wg.Wait()
 }

 func (c providerCache) GetKey(userID *userpb.UserId, spaceID string) string {
@@ -20,10 +20,12 @@ package cache

 import (
 	"strings"
+	"sync"
 	"time"

 	userpb "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
+	"go-micro.dev/v4/store"
 )

 // NewStatCache creates a new StatCache
@@ -50,27 +52,27 @@ func (c statCache) RemoveStat(userID *userpb.UserId, res *provider.ResourceId) {
 		oid = "oid:" + res.OpaqueId
 	}

-	keys, err := c.List()
-	if err != nil {
-		// FIXME handle error
-		return
-	}
-	for _, key := range keys {
-		if strings.Contains(key, uid) {
-			_ = c.Delete(key)
-			continue
-		}
+	// TODO currently, invalidating the stat cache is inefficient and should be disabled. Storage providers / drivers can more selectively invalidate stat cache entries.
+	// This shotgun invalidation wipes all cache entries for the user, space, and nodeid of a changed resource, which means the stat cache is mostly empty, anyway.
+	prefixes := []string{uid, "*" + sid, "*" + oid}

-		if sid != "" && strings.Contains(key, sid) {
-			_ = c.Delete(key)
-			continue
-		}
-
-		if oid != "" && strings.Contains(key, oid) {
-			_ = c.Delete(key)
-			continue
-		}
+	wg := sync.WaitGroup{}
+	for _, prefix := range prefixes {
+		wg.Add(1)
+		go func(p string) {
+			defer wg.Done()
+			keys, _ := c.List(store.ListPrefix(p), store.ListLimit(100))
+			for _, key := range keys {
+				wg.Add(1)
+				go func(k string) {
+					defer wg.Done()
+					_ = c.Delete(k)
+				}(key)
+			}
+		}(prefix)
 	}
+
+	wg.Wait()
 }

 // generates a user specific key pointing to ref - used for statcache
@@ -48,7 +48,6 @@ import (
 	"github.com/cs3org/reva/v2/pkg/storage/utils/chunking"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/lookup"
 	"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/storage/utils/decomposedfs/migrator"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node"
 	"github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/options"
@@ -589,16 +588,6 @@ func (fs *Decomposedfs) CreateDir(ctx context.Context, ref *provider.Reference)
 		return
 	}

-	if fs.o.TreeTimeAccounting || fs.o.TreeSizeAccounting {
-		// mark the home node as the end of propagation
-		if err = n.SetXattrString(prefixes.PropagationAttr, "1"); err != nil {
-			appctx.GetLogger(ctx).Error().Err(err).Interface("node", n).Msg("could not mark node to propagate")
-
-			// FIXME: This does not return an error at all, but results in a severe situation that the
-			// part tree is not marked for propagation
-			return
-		}
-	}
 	return
 }

@@ -57,15 +57,11 @@ func (MessagePackBackend) Name() string { return "messagepack" }

 // All reads all extended attributes for a node
 func (b MessagePackBackend) All(path string) (map[string][]byte, error) {
-	path = b.MetadataPath(path)
-
 	return b.loadAttributes(path, nil)
 }

 // Get an extended attribute value for the given key
 func (b MessagePackBackend) Get(path, key string) ([]byte, error) {
-	path = b.MetadataPath(path)
-
 	attribs, err := b.loadAttributes(path, nil)
 	if err != nil {
 		return []byte{}, err
@@ -79,8 +75,6 @@ func (b MessagePackBackend) Get(path, key string) ([]byte, error) {

 // GetInt64 reads a string as int64 from the xattrs
 func (b MessagePackBackend) GetInt64(path, key string) (int64, error) {
-	path = b.MetadataPath(path)
-
 	attribs, err := b.loadAttributes(path, nil)
 	if err != nil {
 		return 0, err
@@ -99,8 +93,6 @@ func (b MessagePackBackend) GetInt64(path, key string) (int64, error) {
 // List retrieves a list of names of extended attributes associated with the
 // given path in the file system.
 func (b MessagePackBackend) List(path string) ([]string, error) {
-	path = b.MetadataPath(path)
-
 	attribs, err := b.loadAttributes(path, nil)
 	if err != nil {
 		return nil, err
@@ -130,7 +122,6 @@ func (b MessagePackBackend) Remove(path, key string) error {
 // AllWithLockedSource reads all extended attributes from the given reader (if possible).
 // The path argument is used for storing the data in the cache
 func (b MessagePackBackend) AllWithLockedSource(path string, source io.Reader) (map[string][]byte, error) {
-	path = b.MetadataPath(path)
 	return b.loadAttributes(path, source)
 }

@@ -139,11 +130,11 @@ func (b MessagePackBackend) saveAttributes(path string, setAttribs map[string][]
 		f   readWriteCloseSeekTruncater
 		err error
 	)
-	path = b.MetadataPath(path)
+	metaPath := b.MetadataPath(path)
 	if acquireLock {
-		f, err = lockedfile.OpenFile(path, os.O_RDWR|os.O_CREATE, 0600)
+		f, err = lockedfile.OpenFile(metaPath, os.O_RDWR|os.O_CREATE, 0600)
 	} else {
-		f, err = os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0600)
+		f, err = os.OpenFile(metaPath, os.O_RDWR|os.O_CREATE, 0600)
 	}
 	if err != nil {
 		return err
@@ -151,7 +142,7 @@ func (b MessagePackBackend) saveAttributes(path string, setAttribs map[string][]
 	defer f.Close()

 	// Invalidate cache early
-	_ = b.metaCache.RemoveMetadata(path)
+	_ = b.metaCache.RemoveMetadata(b.cacheKey(path))

 	// Read current state
 	msgBytes, err := io.ReadAll(f)
@@ -204,15 +195,16 @@ func (b MessagePackBackend) loadAttributes(path string, source io.Reader) (map[s
 		return attribs, err
 	}

+	metaPath := b.MetadataPath(path)
 	if source == nil {
-		source, err = lockedfile.Open(path)
+		source, err = lockedfile.Open(metaPath)
 		// // No cached entry found. Read from storage and store in cache
 		if err != nil {
 			if os.IsNotExist(err) {
 				// some of the caller rely on ENOTEXISTS to be returned when the
 				// actual file (not the metafile) does not exist in order to
 				// determine whether a node exists or not -> stat the actual node
-				_, err := os.Stat(strings.TrimSuffix(path, ".mpk"))
+				_, err := os.Stat(path)
 				if err != nil {
 					return nil, err
 				}
@@ -254,13 +246,15 @@ func (b MessagePackBackend) Purge(path string) error {

 // Rename moves the data for a given path to a new path
 func (b MessagePackBackend) Rename(oldPath, newPath string) error {
-	data := map[string]string{}
-	_ = b.metaCache.PullFromCache(b.cacheKey(oldPath), &data)
-	err := b.metaCache.RemoveMetadata(b.cacheKey(oldPath))
-	if err != nil {
-		return err
+	data := map[string][]byte{}
+	err := b.metaCache.PullFromCache(b.cacheKey(oldPath), &data)
+	if err == nil {
+		err = b.metaCache.PushToCache(b.cacheKey(newPath), data)
+		if err != nil {
+			return err
+		}
 	}
-	err = b.metaCache.PushToCache(b.cacheKey(newPath), data)
+	err = b.metaCache.RemoveMetadata(b.cacheKey(oldPath))
 	if err != nil {
 		return err
 	}
@@ -164,48 +164,22 @@ func (n *Node) SetType(t provider.ResourceType) {
 	n.nodeType = &t
 }

-// ChangeOwner sets the owner of n to newOwner
-func (n *Node) ChangeOwner(new *userpb.UserId) (err error) {
-	n.SpaceRoot.owner = new
-
-	attribs := Attributes{}
-	attribs.SetString(prefixes.OwnerIDAttr, new.OpaqueId)
-	attribs.SetString(prefixes.OwnerIDPAttr, new.Idp)
-	attribs.SetString(prefixes.OwnerTypeAttr, utils.UserTypeToString(new.Type))
-
-	if err := n.SpaceRoot.SetXattrs(attribs, true); err != nil {
-		return err
-	}
-
-	return
-}
-
-// WriteAllNodeMetadata writes the Node metadata to disk
-func (n *Node) WriteAllNodeMetadata(ctx context.Context) (err error) {
+// NodeMetadata writes the Node metadata to disk and allows passing additional attributes
+func (n *Node) NodeMetadata() Attributes {
 	attribs := Attributes{}
 	attribs.SetInt64(prefixes.TypeAttr, int64(n.Type()))
 	attribs.SetString(prefixes.ParentidAttr, n.ParentID)
 	attribs.SetString(prefixes.NameAttr, n.Name)
-	attribs.SetString(prefixes.BlobIDAttr, n.BlobID)
-	attribs.SetInt64(prefixes.BlobsizeAttr, n.Blobsize)
-
-	return n.SetXattrs(attribs, true)
+	if n.Type() == provider.ResourceType_RESOURCE_TYPE_FILE {
+		attribs.SetString(prefixes.BlobIDAttr, n.BlobID)
+		attribs.SetInt64(prefixes.BlobsizeAttr, n.Blobsize)
+	}
+	return attribs
 }

-// WriteOwner writes the space owner
-func (n *Node) WriteOwner(owner *userpb.UserId) error {
+// SetOwner sets the space owner on the node
+func (n *Node) SetOwner(owner *userpb.UserId) {
 	n.SpaceRoot.owner = owner
-
-	attribs := Attributes{}
-	attribs.SetString(prefixes.OwnerIDAttr, owner.OpaqueId)
-	attribs.SetString(prefixes.OwnerIDPAttr, owner.Idp)
-	attribs.SetString(prefixes.OwnerTypeAttr, utils.UserTypeToString(owner.Type))
-
-	if err := n.SpaceRoot.SetXattrs(attribs, true); err != nil {
-		return err
-	}
-	n.SpaceRoot.owner = owner
-	return nil
 }

 // SpaceOwnerOrManager returns the space owner of the space. If no owner is set
@@ -354,11 +328,13 @@ func ReadNode(ctx context.Context, lu PathLookup, spaceID, nodeID string, canLis

 	if revisionSuffix == "" {
 		n.BlobID = attrs.String(prefixes.BlobIDAttr)
-		blobSize, err := attrs.Int64(prefixes.BlobsizeAttr)
-		if err != nil {
-			return nil, err
+		if n.BlobID != "" {
+			blobSize, err := attrs.Int64(prefixes.BlobsizeAttr)
+			if err != nil {
+				return nil, err
+			}
+			n.Blobsize = blobSize
 		}
-		n.Blobsize = blobSize
 	} else {
 		n.BlobID, err = lu.ReadBlobIDAttr(nodePath + revisionSuffix)
 		if err != nil {
@@ -101,28 +101,18 @@ func (fs *Decomposedfs) CreateStorageSpace(ctx context.Context, req *provider.Cr
 		return nil, errors.Wrap(err, "Decomposedfs: error creating node")
 	}

-	if err := root.WriteAllNodeMetadata(ctx); err != nil {
-		return nil, err
-	}
-	var owner *userv1beta1.UserId
 	if req.GetOwner() != nil && req.GetOwner().GetId() != nil {
-		owner = req.GetOwner().GetId()
+		root.SetOwner(req.GetOwner().GetId())
 	} else {
-		owner = &userv1beta1.UserId{OpaqueId: spaceID, Type: userv1beta1.UserType_USER_TYPE_SPACE_OWNER}
-	}
-	if err := root.WriteOwner(owner); err != nil {
-		return nil, err
+		root.SetOwner(&userv1beta1.UserId{OpaqueId: spaceID, Type: userv1beta1.UserType_USER_TYPE_SPACE_OWNER})
 	}

-	err = fs.updateIndexes(ctx, req.GetOwner().GetId().GetOpaqueId(), req.Type, root.ID)
-	if err != nil {
-		return nil, err
-	}
+	metadata := node.Attributes{}
+	metadata.SetString(prefixes.OwnerIDAttr, root.Owner().GetOpaqueId())
+	metadata.SetString(prefixes.OwnerIDPAttr, root.Owner().GetIdp())
+	metadata.SetString(prefixes.OwnerTypeAttr, utils.UserTypeToString(root.Owner().GetType()))

-	metadata := make(node.Attributes, 6)
-
-	// always enable propagation on the storage space root
-	// mark the space root node as the end of propagation
+	// always mark the space root node as the end of propagation
 	metadata.SetString(prefixes.PropagationAttr, "1")
 	metadata.SetString(prefixes.NameAttr, req.Name)
 	metadata.SetString(prefixes.SpaceNameAttr, req.Name)
@@ -151,14 +141,20 @@ func (fs *Decomposedfs) CreateStorageSpace(ctx context.Context, req *provider.Cr
 		metadata.SetString(prefixes.SpaceAliasAttr, alias)
 	}

+	// Write node
 	if err := root.SetXattrs(metadata, true); err != nil {
 		return nil, err
 	}

+	// Write index
+	err = fs.updateIndexes(ctx, req.GetOwner().GetId().GetOpaqueId(), req.Type, root.ID)
+	if err != nil {
+		return nil, err
+	}
+
 	ctx = context.WithValue(ctx, utils.SpaceGrant, struct{ SpaceType string }{SpaceType: req.Type})

 	if req.Type != _spaceTypePersonal {
-		u := ctxpkg.ContextMustGetUser(ctx)
 		if err := fs.AddGrant(ctx, &provider.Reference{
 			ResourceId: &provider.ResourceId{
 				SpaceId:  spaceID,
@@ -633,12 +629,17 @@ func (fs *Decomposedfs) DeleteStorageSpace(ctx context.Context, req *provider.De
 			return err
 		}

+		// invalidate cache
+		if err := fs.lu.MetadataBackend().Purge(n.InternalPath()); err != nil {
+			return err
+		}
+
 		// remove space metadata
 		if err := os.RemoveAll(fs.getSpaceRoot(spaceID)); err != nil {
 			return err
 		}

-		// FIXME remove space blobs
+		// TODO remove space blobs with s3 backend by adding a purge method to the Blobstore interface

 		return nil
 	}
@@ -151,15 +151,15 @@ func (t *Tree) TouchFile(ctx context.Context, n *node.Node, markprocessing bool)
 		return errors.Wrap(err, "Decomposedfs: error creating node")
 	}

-	err = n.WriteAllNodeMetadata(ctx)
+	attributes := n.NodeMetadata()
+	if markprocessing {
+		attributes[prefixes.StatusPrefix] = []byte(node.ProcessingStatus)
+	}
+	err = n.SetXattrs(attributes, true)
 	if err != nil {
 		return err
 	}

-	if markprocessing {
-		_ = n.SetXattr(prefixes.StatusPrefix, []byte(node.ProcessingStatus))
-	}
-
 	// link child name to parent if it is new
 	childNameLink := filepath.Join(n.ParentPath(), n.Name)
 	var link string
@@ -196,10 +196,6 @@ func (t *Tree) CreateDir(ctx context.Context, n *node.Node) (err error) {
 		return
 	}

-	if err := n.SetTreeSize(0); err != nil {
-		return err
-	}
-
 	// make child appear in listings
 	relativeNodePath := filepath.Join("../../../../../", lookup.Pathify(n.ID, 4, 2))
 	err = os.Symlink(relativeNodePath, filepath.Join(n.ParentPath(), n.Name))
@@ -907,7 +903,12 @@ func (t *Tree) createDirNode(ctx context.Context, n *node.Node) (err error) {
 		return errors.Wrap(err, "Decomposedfs: error creating node")
 	}

-	return n.WriteAllNodeMetadata(ctx)
+	attributes := n.NodeMetadata()
+	attributes[prefixes.TreesizeAttr] = []byte("0") // initialize as empty, TODO why bother? if it is not set we could treat it as 0?
+	if t.options.TreeTimeAccounting || t.options.TreeSizeAccounting {
+		attributes[prefixes.PropagationAttr] = []byte("1") // mark the node for propagation
+	}
+	return n.SetXattrs(attributes, true)
 }

 var nodeIDRegep = regexp.MustCompile(`.*/nodes/([^.]*).*`)
@@ -938,19 +939,21 @@ func (t *Tree) readRecycleItem(ctx context.Context, spaceID, key, path string) (
 	if err != nil {
 		return
 	}
-	recycleNode.SetType(t.lookup.TypeFromPath(recycleNode.InternalPath()))
+	recycleNode.SetType(t.lookup.TypeFromPath(deletedNodePath))

 	var attrBytes []byte
-	// lookup blobID in extended attributes
-	if attrBytes, err = backend.Get(deletedNodePath, prefixes.BlobIDAttr); err == nil {
-		recycleNode.BlobID = string(attrBytes)
-	} else {
-		return
-	}
+	if recycleNode.Type() == provider.ResourceType_RESOURCE_TYPE_FILE {
+		// lookup blobID in extended attributes
+		if attrBytes, err = backend.Get(deletedNodePath, prefixes.BlobIDAttr); err == nil {
+			recycleNode.BlobID = string(attrBytes)
+		} else {
+			return
+		}

-	// lookup blobSize in extended attributes
-	if recycleNode.Blobsize, err = backend.GetInt64(deletedNodePath, prefixes.BlobsizeAttr); err != nil {
-		return
+		// lookup blobSize in extended attributes
+		if recycleNode.Blobsize, err = backend.GetInt64(deletedNodePath, prefixes.BlobsizeAttr); err != nil {
+			return
+		}
 	}

 	// lookup parent id in extended attributes