chore:reva bump v.2.32 (#737)

vendor/github.com/opencloud-eu/reva/v2/internal/http/services/owncloud/ocdav/copy.go

@@ -56,7 +56,7 @@ func (s *svc) handlePathCopy(w http.ResponseWriter, r *http.Request, ns string)
 	ctx, span := appctx.GetTracerProvider(r.Context()).Tracer(tracerName).Start(r.Context(), "copy")
 	defer span.End()
 
-	if r.Body != http.NoBody {
+	if !isBodyEmpty(r) {
 		w.WriteHeader(http.StatusUnsupportedMediaType)
 		b, err := errors.Marshal(http.StatusUnsupportedMediaType, "body must be empty", "", "")
 		errors.HandleWebdavError(appctx.GetLogger(ctx), w, b, err)
@@ -331,7 +331,7 @@ func (s *svc) handleSpacesCopy(w http.ResponseWriter, r *http.Request, spaceID s
 	ctx, span := appctx.GetTracerProvider(r.Context()).Tracer(tracerName).Start(r.Context(), "spaces_copy")
 	defer span.End()
 
-	if r.Body != http.NoBody {
+	if !isBodyEmpty(r) {
 		w.WriteHeader(http.StatusUnsupportedMediaType)
 		b, err := errors.Marshal(http.StatusUnsupportedMediaType, "body must be empty", "", "")
 		errors.HandleWebdavError(appctx.GetLogger(ctx), w, b, err)

vendor/github.com/opencloud-eu/reva/v2/internal/http/services/owncloud/ocdav/delete.go

@@ -39,7 +39,7 @@ func (s *svc) handlePathDelete(w http.ResponseWriter, r *http.Request, ns string
 	ctx, span := appctx.GetTracerProvider(r.Context()).Tracer(tracerName).Start(ctx, "path_delete")
 	defer span.End()
 
-	if r.Body != http.NoBody {
+	if !isBodyEmpty(r) {
 		return http.StatusUnsupportedMediaType, errors.New("body must be empty")
 	}
 
@@ -126,7 +126,7 @@ func (s *svc) handleSpacesDelete(w http.ResponseWriter, r *http.Request, spaceID
 	ctx, span := appctx.GetTracerProvider(r.Context()).Tracer(tracerName).Start(ctx, "spaces_delete")
 	defer span.End()
 
-	if r.Body != http.NoBody {
+	if !isBodyEmpty(r) {
 		return http.StatusUnsupportedMediaType, errors.New("body must be empty")
 	}
 

vendor/github.com/opencloud-eu/reva/v2/internal/http/services/owncloud/ocdav/mkcol.go

@@ -107,7 +107,7 @@ func (s *svc) handleSpacesMkCol(w http.ResponseWriter, r *http.Request, spaceID
 }
 
 func (s *svc) handleMkcol(ctx context.Context, w http.ResponseWriter, r *http.Request, parentRef, childRef *provider.Reference, log zerolog.Logger) (status int, err error) {
-	if r.Body != http.NoBody {
+	if !isBodyEmpty(r) {
 		// We currently do not support extended mkcol https://datatracker.ietf.org/doc/rfc5689/
 		// TODO let clients send a body with properties to set on the new resource
 		return http.StatusUnsupportedMediaType, fmt.Errorf("extended-mkcol not supported")

vendor/github.com/opencloud-eu/reva/v2/internal/http/services/owncloud/ocdav/move.go

@@ -42,7 +42,7 @@ func (s *svc) handlePathMove(w http.ResponseWriter, r *http.Request, ns string)
 	ctx, span := appctx.GetTracerProvider(r.Context()).Tracer(tracerName).Start(r.Context(), "move")
 	defer span.End()
 
-	if r.Body != http.NoBody {
+	if !isBodyEmpty(r) {
 		w.WriteHeader(http.StatusUnsupportedMediaType)
 		b, err := errors.Marshal(http.StatusUnsupportedMediaType, "body must be empty", "", "")
 		errors.HandleWebdavError(appctx.GetLogger(ctx), w, b, err)
@@ -106,7 +106,7 @@ func (s *svc) handleSpacesMove(w http.ResponseWriter, r *http.Request, srcSpaceI
 	ctx, span := appctx.GetTracerProvider(r.Context()).Tracer(tracerName).Start(r.Context(), "spaces_move")
 	defer span.End()
 
-	if r.Body != http.NoBody {
+	if !isBodyEmpty(r) {
 		w.WriteHeader(http.StatusUnsupportedMediaType)
 		b, err := errors.Marshal(http.StatusUnsupportedMediaType, "body must be empty", "", "")
 		errors.HandleWebdavError(appctx.GetLogger(ctx), w, b, err)

vendor/github.com/opencloud-eu/reva/v2/internal/http/services/owncloud/ocdav/ocdav.go

@@ -20,6 +20,7 @@ package ocdav
 
 import (
 	"context"
+	"io"
 	"net/http"
 	"path"
 	"strings"
@@ -399,3 +400,17 @@ func (s *svc) referenceIsChildOf(ctx context.Context, selector pool.Selectable[g
 func filename(p string) string {
 	return strings.Trim(path.Base(p), "/")
 }
+
+// isBodyEmpty returns true when the Body of the request is Empty
+func isBodyEmpty(r *http.Request) bool {
+	if r.Body != nil && r.Body != http.NoBody {
+		buf := make([]byte, 0)
+		_, err := r.Body.Read(buf)
+		if err != io.EOF {
+			// We currently do not support extended mkcol https://datatracker.ietf.org/doc/rfc5689/
+			// TODO let clients send a body with properties to set on the new resource
+			return false
+		}
+	}
+	return true
+}

vendor/github.com/opencloud-eu/reva/v2/pkg/storage/fs/posix/lookup/lookup.go

@@ -399,11 +399,14 @@ func (lu *Lookup) GenerateSpaceID(spaceType string, owner *user.User) (string, e
 	case _spaceTypeProject:
 		return uuid.New().String(), nil
 	case _spaceTypePersonal:
-		path := templates.WithUser(owner, lu.Options.PersonalSpacePathTemplate)
+		relPath := templates.WithUser(owner, lu.Options.PersonalSpacePathTemplate)
+		path := filepath.Join(lu.Options.Root, relPath)
 
-		spaceID, _, err := lu.IDsForPath(context.TODO(), filepath.Join(lu.Options.Root, path))
+		// do we already know about this space?
+		spaceID, _, err := lu.IDsForPath(context.TODO(), path)
 		if err != nil {
-			_, err := os.Stat(filepath.Join(lu.Options.Root, path))
+			// check if the space exists on disk incl. attributes
+			spaceID, _, _, _, err := lu.metadataBackend.IdentifyPath(context.TODO(), path)
 			if err != nil {
 				if metadata.IsNotExist(err) || metadata.IsAttrUnset(err) {
 					return uuid.New().String(), nil
@@ -411,6 +414,10 @@ func (lu *Lookup) GenerateSpaceID(spaceType string, owner *user.User) (string, e
 					return "", err
 				}
 			}
+			if len(spaceID) == 0 {
+				return "", errtypes.InternalError("encountered empty space id on disk")
+			}
+			return spaceID, nil
 		}
 		return spaceID, nil
 	default:

vendor/github.com/opencloud-eu/reva/v2/pkg/storage/fs/posix/options/options.go

@@ -39,11 +39,12 @@ type Options struct {
 	// a revision when the file is changed.
 	EnableFSRevisions bool `mapstructure:"enable_fs_revisions"`
 
-	ScanFS                  bool   `mapstructure:"scan_fs"`
-	WatchFS                 bool   `mapstructure:"watch_fs"`
-	WatchType               string `mapstructure:"watch_type"`
-	WatchPath               string `mapstructure:"watch_path"`
-	WatchFolderKafkaBrokers string `mapstructure:"watch_folder_kafka_brokers"`
+	ScanFS                   bool   `mapstructure:"scan_fs"`
+	WatchFS                  bool   `mapstructure:"watch_fs"`
+	WatchType                string `mapstructure:"watch_type"`
+	WatchPath                string `mapstructure:"watch_path"`
+	WatchRoot                string `mapstructure:"watch_root"` // base directory for the watch. events will be considered relative to this path
+	WatchNotificationBrokers string `mapstructure:"watch_notification_brokers"`
 
 	// InotifyWatcher specific options
 	InotifyStatsFrequency time.Duration `mapstructure:"inotify_stats_frequency"`

vendor/github.com/opencloud-eu/reva/v2/pkg/storage/fs/posix/tree/assimilation.go

@@ -36,7 +36,6 @@ import (
 	"github.com/pkg/xattr"
 	"github.com/rs/zerolog/log"
 
-	userv1beta1 "github.com/cs3org/go-cs3apis/cs3/identity/user/v1beta1"
 	provider "github.com/cs3org/go-cs3apis/cs3/storage/provider/v1beta1"
 	"github.com/opencloud-eu/reva/v2/pkg/events"
 	"github.com/opencloud-eu/reva/v2/pkg/storage/pkg/decomposedfs/metadata"
@@ -342,10 +341,9 @@ func (t *Tree) getNodeForPath(path string) (*node.Node, error) {
 	return node.ReadNode(context.Background(), t.lookup, spaceID, nodeID, false, nil, false)
 }
 
-func (t *Tree) findSpaceId(path string) (string, node.Attributes, error) {
+func (t *Tree) findSpaceId(path string) (string, error) {
 	// find the space id, scope by the according user
 	spaceCandidate := path
-	spaceAttrs := node.Attributes{}
 	for strings.HasPrefix(spaceCandidate, t.options.Root) {
 		spaceID, _, err := t.lookup.IDsForPath(context.Background(), spaceCandidate)
 		if err == nil && len(spaceID) > 0 {
@@ -353,67 +351,62 @@ func (t *Tree) findSpaceId(path string) (string, node.Attributes, error) {
 				// set the uid and gid for the space
 				fi, err := os.Stat(spaceCandidate)
 				if err != nil {
-					return "", spaceAttrs, err
+					return "", err
 				}
 				sys := fi.Sys().(*syscall.Stat_t)
 				gid := int(sys.Gid)
 				_, err = t.userMapper.ScopeUserByIds(-1, gid)
 				if err != nil {
-					return "", spaceAttrs, err
+					return "", err
 				}
 			}
 
-			return spaceID, spaceAttrs, nil
+			return spaceID, nil
 		}
 		spaceCandidate = filepath.Dir(spaceCandidate)
 	}
-	return "", spaceAttrs, fmt.Errorf("could not find space for path %s", path)
+	return "", fmt.Errorf("could not find space for path %s", path)
 }
 
 func (t *Tree) assimilate(item scanItem) error {
 	t.log.Debug().Str("path", item.Path).Bool("rescan", item.ForceRescan).Bool("recurse", item.Recurse).Msg("assimilate")
 	var err error
 
-	// First find the space id
-	spaceID, spaceAttrs, err := t.findSpaceId(item.Path)
+	spaceID, id, parentID, mtime, err := t.lookup.MetadataBackend().IdentifyPath(context.Background(), item.Path)
 	if err != nil {
 		return err
 	}
 
-	assimilationNode := &assimilationNode{
-		spaceID: spaceID,
-		path:    item.Path,
-	}
-
-	// lock the file for assimilation
-	unlock, err := t.lookup.MetadataBackend().Lock(assimilationNode)
-	if err != nil {
-		return errors.Wrap(err, "failed to lock item for assimilation")
-	}
-	defer func() {
-		_ = unlock()
-	}()
-
-	user := &userv1beta1.UserId{
-		Idp:      string(spaceAttrs[prefixes.OwnerIDPAttr]),
-		OpaqueId: string(spaceAttrs[prefixes.OwnerIDAttr]),
-	}
-
-	// check for the id attribute again after grabbing the lock, maybe the file was assimilated/created by us in the meantime
-	_, id, parentID, mtime, err := t.lookup.MetadataBackend().IdentifyPath(context.Background(), item.Path)
-	if err != nil {
-		return err
+	if spaceID == "" {
+		// node didn't have a space ID attached. try to find it by walking up the path on disk
+		spaceID, err = t.findSpaceId(item.Path)
+		if err != nil {
+			return err
+		}
 	}
 
 	if id != "" {
 		// the file has an id set, we already know it from the past
-		// n := node.NewBaseNode(spaceID, id, t.lookup)
+
+		// lock the file for re-assimilation
+		assimilationNode := &assimilationNode{
+			spaceID: spaceID,
+			nodeId:  id,
+			path:    item.Path,
+		}
+
+		unlock, err := t.lookup.MetadataBackend().Lock(assimilationNode)
+		if err != nil {
+			return errors.Wrap(err, "failed to lock item for assimilation")
+		}
+		defer func() {
+			_ = unlock()
+		}()
 
 		previousPath, ok := t.lookup.GetCachedID(context.Background(), spaceID, id)
 		if previousPath == "" || !ok {
 			previousPath, ok = t.lookup.IDHistoryCache.Get(context.Background(), spaceID, id)
 		}
-		// previousParentID, _ := t.lookup.MetadataBackend().Get(context.Background(), n, prefixes.ParentidAttr)
 
 		// compare metadata mtime with actual mtime. if it matches AND the path hasn't changed (move operation)
 		// we can skip the assimilation because the file was handled by us
@@ -446,7 +439,7 @@ func (t *Tree) assimilate(item scanItem) error {
 				if err := t.lookup.CacheID(context.Background(), spaceID, id, item.Path); err != nil {
 					t.log.Error().Err(err).Str("spaceID", spaceID).Str("id", id).Str("path", item.Path).Msg("could not cache id")
 				}
-				_, attrs, err := t.updateFile(item.Path, id, spaceID)
+				_, attrs, err := t.updateFile(item.Path, id, spaceID, fi)
 				if err != nil {
 					return err
 				}
@@ -484,9 +477,6 @@ func (t *Tree) assimilate(item scanItem) error {
 						Path: filepath.Base(previousPath),
 					}
 					t.PublishEvent(events.ItemMoved{
-						SpaceOwner:   user,
-						Executant:    user,
-						Owner:        user,
 						Ref:          ref,
 						OldReference: oldRef,
 						Timestamp:    utils.TSNow(),
@@ -500,16 +490,38 @@ func (t *Tree) assimilate(item scanItem) error {
 				t.log.Error().Err(err).Str("spaceID", spaceID).Str("id", id).Str("path", item.Path).Msg("could not cache id")
 			}
 
-			_, _, err := t.updateFile(item.Path, id, spaceID)
+			_, _, err := t.updateFile(item.Path, id, spaceID, fi)
 			if err != nil {
 				return err
 			}
 		}
 	} else {
 		t.log.Debug().Str("path", item.Path).Msg("new item detected")
+		assimilationNode := &assimilationNode{
+			spaceID: spaceID,
+			// Use the path as the node ID (which is used for calculating the lock file path) since we do not have an ID yet
+			nodeId: strings.ReplaceAll(strings.TrimPrefix(item.Path, "/"), "/", "-"),
+		}
+		unlock, err := t.lookup.MetadataBackend().Lock(assimilationNode)
+		if err != nil {
+			return err
+		}
+		defer func() { _ = unlock() }()
+
+		// check if the file got an ID while we were waiting for the lock
+		_, id, _, _, err = t.lookup.MetadataBackend().IdentifyPath(context.Background(), item.Path)
+		if err != nil {
+			return err
+		}
+		if id != "" {
+			// file was assimilated by another thread while we were waiting for the lock
+			t.log.Debug().Str("path", item.Path).Msg("file was assimilated by another thread")
+			return nil
+		}
+
 		// assimilate new file
 		newId := uuid.New().String()
-		fi, _, err := t.updateFile(item.Path, newId, spaceID)
+		fi, _, err := t.updateFile(item.Path, newId, spaceID, nil)
 		if err != nil {
 			return err
 		}
@@ -523,25 +535,19 @@ func (t *Tree) assimilate(item scanItem) error {
 		}
 		if fi.IsDir() {
 			t.PublishEvent(events.ContainerCreated{
-				SpaceOwner: user,
-				Executant:  user,
-				Owner:      user,
-				Ref:        ref,
-				Timestamp:  utils.TSNow(),
+				Ref:       ref,
+				Timestamp: utils.TSNow(),
 			})
 		} else {
 			if fi.Size() == 0 {
 				t.PublishEvent(events.FileTouched{
-					SpaceOwner: user,
-					Executant:  user,
-					Ref:        ref,
-					Timestamp:  utils.TSNow(),
+					Ref:       ref,
+					Timestamp: utils.TSNow(),
 				})
 			} else {
 				t.PublishEvent(events.UploadReady{
-					SpaceOwner: user,
-					FileRef:    ref,
-					Timestamp:  utils.TSNow(),
+					FileRef:   ref,
+					Timestamp: utils.TSNow(),
 				})
 			}
 		}
@@ -549,7 +555,7 @@ func (t *Tree) assimilate(item scanItem) error {
 	return nil
 }
 
-func (t *Tree) updateFile(path, id, spaceID string) (fs.FileInfo, node.Attributes, error) {
+func (t *Tree) updateFile(path, id, spaceID string, fi fs.FileInfo) (fs.FileInfo, node.Attributes, error) {
 	retries := 1
 	parentID := ""
 	bn := assimilationNode{spaceID: spaceID, nodeId: id, path: path}
@@ -585,9 +591,12 @@ assimilate:
 	}
 
 	// assimilate file
-	fi, err := os.Stat(path)
-	if err != nil {
-		return nil, nil, errors.Wrap(err, "failed to stat item")
+	if fi == nil {
+		var err error
+		fi, err = os.Stat(path)
+		if err != nil {
+			return nil, nil, errors.Wrap(err, "failed to stat item")
+		}
 	}
 
 	attrs, err := t.lookup.MetadataBackend().All(context.Background(), bn)
@@ -604,13 +613,6 @@ assimilate:
 		attributes[prefixes.ParentidAttr] = []byte(parentID)
 	}
 
-	sha1h, md5h, adler32h, err := node.CalculateChecksums(context.Background(), path)
-	if err == nil {
-		attributes[prefixes.ChecksumPrefix+"sha1"] = sha1h.Sum(nil)
-		attributes[prefixes.ChecksumPrefix+"md5"] = md5h.Sum(nil)
-		attributes[prefixes.ChecksumPrefix+"adler32"] = adler32h.Sum(nil)
-	}
-
 	var n *node.Node
 	if fi.IsDir() {
 		attributes.SetInt64(prefixes.TypeAttr, int64(provider.ResourceType_RESOURCE_TYPE_CONTAINER))
@@ -625,6 +627,13 @@ assimilate:
 		}
 		n = node.New(spaceID, id, parentID, filepath.Base(path), treeSize, "", provider.ResourceType_RESOURCE_TYPE_CONTAINER, nil, t.lookup)
 	} else {
+		sha1h, md5h, adler32h, err := node.CalculateChecksums(context.Background(), path)
+		if err == nil {
+			attributes[prefixes.ChecksumPrefix+"sha1"] = sha1h.Sum(nil)
+			attributes[prefixes.ChecksumPrefix+"md5"] = md5h.Sum(nil)
+			attributes[prefixes.ChecksumPrefix+"adler32"] = adler32h.Sum(nil)
+		}
+
 		blobID := uuid.NewString()
 		attributes.SetString(prefixes.BlobIDAttr, blobID)
 		attributes.SetInt64(prefixes.BlobsizeAttr, fi.Size())

vendor/github.com/opencloud-eu/reva/v2/pkg/storage/fs/posix/tree/cephfswatcher.go

@@ -0,0 +1,120 @@
+// Copyright 2025 OpenCloud GmbH <mail@opencloud.eu>
+// SPDX-License-Identifier: Apache-2.0
+
+package tree
+
+import (
+	"context"
+	"encoding/json"
+	"path/filepath"
+
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
+	kafka "github.com/segmentio/kafka-go"
+)
+
+const (
+	CEPH_MDS_NOTIFY_ACCESS        = 0x0000000000000001 // 1
+	CEPH_MDS_NOTIFY_ATTRIB        = 0x0000000000000002 // 2
+	CEPH_MDS_NOTIFY_CLOSE_WRITE   = 0x0000000000000004 // 4
+	CEPH_MDS_NOTIFY_CLOSE_NOWRITE = 0x0000000000000008 // 8
+	CEPH_MDS_NOTIFY_CREATE        = 0x0000000000000010 // 16
+	CEPH_MDS_NOTIFY_DELETE        = 0x0000000000000020 // 32
+	CEPH_MDS_NOTIFY_DELETE_SELF   = 0x0000000000000040 // 64
+	CEPH_MDS_NOTIFY_MODIFY        = 0x0000000000000080 // 128
+	CEPH_MDS_NOTIFY_MOVE_SELF     = 0x0000000000000100 // 256
+	CEPH_MDS_NOTIFY_MOVED_FROM    = 0x0000000000000200 // 512
+	CEPH_MDS_NOTIFY_MOVED_TO      = 0x0000000000000400 // 1024
+	CEPH_MDS_NOTIFY_OPEN          = 0x0000000000000800 // 2048
+	CEPH_MDS_NOTIFY_CLOSE         = 0x0000000000001000 // 4096
+	CEPH_MDS_NOTIFY_MOVE          = 0x0000000000002000 // 8192
+	CEPH_MDS_NOTIFY_ONESHOT       = 0x0000000000004000 // 16384
+	CEPH_MDS_NOTIFY_IGNORED       = 0x0000000000008000 // 32768
+	CEPH_MDS_NOTIFY_ONLYDIR       = 0x0000000000010000 // 65536
+)
+
+type CephFSWatcher struct {
+	tree    *Tree
+	root    string
+	brokers []string
+	log     *zerolog.Logger
+}
+
+func NewCephfsWatcher(tree *Tree, brokers []string, log *zerolog.Logger) (*CephFSWatcher, error) {
+	return &CephFSWatcher{
+		tree:    tree,
+		root:    tree.options.WatchRoot,
+		brokers: brokers,
+		log:     log,
+	}, nil
+}
+
+type cephfsEvent struct {
+	// Mask/Path are the event mask and path of the according entity
+	Mask int    `json:"mask"`
+	Path string `json:"path"`
+
+	// Src*/Dst* are emitted for the source and destination of move events
+	SrcMask  int    `json:"src_mask"`
+	SrcPath  string `json:"src_path"`
+	DestMask int    `json:"dest_mask"`
+	DestPath string `json:"dest_path"`
+}
+
+func (w *CephFSWatcher) Watch(topic string) {
+	w.log.Info().Str("topic", topic).Msg("cephfs watcher watching topic")
+	r := kafka.NewReader(kafka.ReaderConfig{
+		Brokers: w.brokers,
+		GroupID: "opencloud-posixfs",
+		Topic:   topic,
+	})
+
+	for {
+		m, err := r.ReadMessage(context.Background())
+		if err != nil {
+			log.Error().Err(err).Msg("error reading message")
+			continue
+		}
+
+		ev := &cephfsEvent{}
+		err = json.Unmarshal(m.Value, ev)
+		if err != nil {
+			w.log.Error().Err(err).Msg("error unmarshalling message")
+			continue
+		}
+
+		if w.tree.isIgnored(ev.Path) {
+			continue
+		}
+
+		mask := ev.Mask
+		path := filepath.Join(w.tree.options.WatchRoot, ev.Path)
+		if ev.DestMask > 0 {
+			mask = ev.DestMask
+			path = filepath.Join(w.tree.options.WatchRoot, ev.DestPath)
+		}
+		isDir := mask&CEPH_MDS_NOTIFY_ONLYDIR > 0
+		go func() {
+			switch {
+			case mask&CEPH_MDS_NOTIFY_DELETE > 0:
+				err = w.tree.Scan(path, ActionDelete, isDir)
+			case mask&CEPH_MDS_NOTIFY_CREATE > 0 || mask&CEPH_MDS_NOTIFY_MOVED_TO > 0:
+				if ev.SrcMask > 0 {
+					// This is a move, clean up the old path
+					err = w.tree.Scan(filepath.Join(w.tree.options.WatchRoot, ev.SrcPath), ActionMoveFrom, isDir)
+				}
+				err = w.tree.Scan(path, ActionCreate, isDir)
+			case mask&CEPH_MDS_NOTIFY_CLOSE_WRITE > 0:
+				err = w.tree.Scan(path, ActionUpdate, isDir)
+			case mask&CEPH_MDS_NOTIFY_CLOSE > 0:
+				// ignore, already handled by CLOSE_WRITE
+			default:
+				w.log.Trace().Interface("event", ev).Msg("unhandled event")
+				return
+			}
+			if err != nil {
+				w.log.Error().Err(err).Str("path", path).Msg("error scanning file")
+			}
+		}()
+	}
+}

vendor/github.com/opencloud-eu/reva/v2/pkg/storage/fs/posix/tree/gpfswatchfolderwatcher.go

@@ -22,6 +22,7 @@ import (
 	"context"
 	"encoding/json"
 	"log"
+	"path/filepath"
 	"strconv"
 	"strings"
 
@@ -30,16 +31,18 @@ import (
 )
 
 type GpfsWatchFolderWatcher struct {
-	tree    *Tree
-	brokers []string
-	log     *zerolog.Logger
+	tree       *Tree
+	brokers    []string
+	log        *zerolog.Logger
+	watch_root string
 }
 
 func NewGpfsWatchFolderWatcher(tree *Tree, kafkaBrokers []string, log *zerolog.Logger) (*GpfsWatchFolderWatcher, error) {
 	return &GpfsWatchFolderWatcher{
-		tree:    tree,
-		brokers: kafkaBrokers,
-		log:     log,
+		tree:       tree,
+		brokers:    kafkaBrokers,
+		watch_root: tree.options.WatchRoot,
+		log:        log,
 	}, nil
 }
 
@@ -66,30 +69,32 @@ func (w *GpfsWatchFolderWatcher) Watch(topic string) {
 			continue
 		}
 
+		path := filepath.Join(w.watch_root, lwev.Path)
+
 		go func() {
 			isDir := strings.Contains(lwev.Event, "IN_ISDIR")
 
 			var err error
 			switch {
 			case strings.Contains(lwev.Event, "IN_DELETE"):
-				err = w.tree.Scan(lwev.Path, ActionDelete, isDir)
+				err = w.tree.Scan(path, ActionDelete, isDir)
 
 			case strings.Contains(lwev.Event, "IN_MOVE_FROM"):
-				err = w.tree.Scan(lwev.Path, ActionMoveFrom, isDir)
+				err = w.tree.Scan(path, ActionMoveFrom, isDir)
 
 			case strings.Contains(lwev.Event, "IN_CREATE"):
-				err = w.tree.Scan(lwev.Path, ActionCreate, isDir)
+				err = w.tree.Scan(path, ActionCreate, isDir)
 
 			case strings.Contains(lwev.Event, "IN_CLOSE_WRITE"):
 				bytesWritten, convErr := strconv.Atoi(lwev.BytesWritten)
 				if convErr == nil && bytesWritten > 0 {
-					err = w.tree.Scan(lwev.Path, ActionUpdate, isDir)
+					err = w.tree.Scan(path, ActionUpdate, isDir)
 				}
 			case strings.Contains(lwev.Event, "IN_MOVED_TO"):
-				err = w.tree.Scan(lwev.Path, ActionMove, isDir)
+				err = w.tree.Scan(path, ActionMove, isDir)
 			}
 			if err != nil {
-				w.log.Error().Err(err).Str("path", lwev.Path).Msg("error scanning path")
+				w.log.Error().Err(err).Str("path", path).Msg("error scanning path")
 			}
 		}()
 	}

vendor/github.com/opencloud-eu/reva/v2/pkg/storage/fs/posix/tree/tree.go

@@ -117,9 +117,12 @@ func New(lu node.PathLookup, bs node.Blobstore, um usermapper.Mapper, trashbin *
 	if o.WatchFS {
 		watchPath := o.WatchPath
 		var err error
+
+		t.log.Info().Str("watch type", o.WatchType).Str("path", o.WatchPath).Str("root", o.WatchRoot).
+			Str("brokers", o.WatchNotificationBrokers).Msg("Watching fs")
 		switch o.WatchType {
 		case "gpfswatchfolder":
-			t.watcher, err = NewGpfsWatchFolderWatcher(t, strings.Split(o.WatchFolderKafkaBrokers, ","), log)
+			t.watcher, err = NewGpfsWatchFolderWatcher(t, strings.Split(o.WatchNotificationBrokers, ","), log)
 			if err != nil {
 				return nil, err
 			}
@@ -128,6 +131,11 @@ func New(lu node.PathLookup, bs node.Blobstore, um usermapper.Mapper, trashbin *
 			if err != nil {
 				return nil, err
 			}
+		case "cephfs":
+			t.watcher, err = NewCephfsWatcher(t, strings.Split(o.WatchNotificationBrokers, ","), log)
+			if err != nil {
+				return nil, err
+			}
 		default:
 			t.watcher, err = NewInotifyWatcher(t, o, log)
 			if err != nil {

vendor/github.com/opencloud-eu/reva/v2/pkg/storage/pkg/decomposedfs/metadata/hybrid_backend.go

@@ -46,7 +46,12 @@ func (HybridBackend) Name() string { return "hybrid" }
 
 // IdentifyPath returns the space id, node id and mtime of a file
 func (b HybridBackend) IdentifyPath(_ context.Context, path string) (string, string, string, time.Time, error) {
-	spaceID, _ := xattr.Get(path, prefixes.SpaceIDAttr)
+	spaceID, err := xattr.Get(path, prefixes.SpaceIDAttr)
+	if err != nil {
+		if IsNotExist(err) {
+			return "", "", "", time.Time{}, err
+		}
+	}
 	id, _ := xattr.Get(path, prefixes.IDAttr)
 	parentID, _ := xattr.Get(path, prefixes.ParentidAttr)
 

vendor/github.com/opencloud-eu/reva/v2/pkg/storage/pkg/decomposedfs/upload/upload.go

@@ -292,6 +292,13 @@ func (session *DecomposedFsSession) Finalize(ctx context.Context) (err error) {
 	revisionNode := node.New(session.SpaceID(), session.NodeID(), "", "", session.Size(), session.ID(),
 		provider.ResourceType_RESOURCE_TYPE_FILE, session.SpaceOwner(), session.store.lu)
 
+	// lock the node before writing the blob
+	unlock, err := session.store.lu.MetadataBackend().Lock(revisionNode)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = unlock() }()
+
 	// upload the data to the blobstore
 	_, subspan := tracer.Start(ctx, "WriteBlob")
 	err = session.store.tp.WriteBlob(revisionNode, session.binPath())