diff --git a/pkg/storage/fs/posix/idcache/idcache.go b/pkg/storage/fs/posix/idcache/idcache.go index 91967f76d2e..9c7e202ca71 100644 --- a/pkg/storage/fs/posix/idcache/idcache.go +++ b/pkg/storage/fs/posix/idcache/idcache.go @@ -30,8 +30,20 @@ import ( "github.com/nats-io/nats.go/jetstream" "github.com/opencloud-eu/reva/v2/pkg/appctx" "github.com/opencloud-eu/reva/v2/pkg/errtypes" + "golang.org/x/sync/errgroup" ) +// movePathConcurrency bounds the number of in-flight re-key operations when +// moving a subtree. The work is network bound (NATS KV round-trips), so a +// generous limit keeps the connection busy without overwhelming the server. +const movePathConcurrency = 32 + +type moveEntry struct { + reverseKey string + spaceID string + nodeID string + currentPath string +} type IDCache struct { kv jetstream.KeyValue } @@ -129,6 +141,92 @@ func (c *IDCache) DeletePath(ctx context.Context, path string) error { }) } +// MovePath recursively re-keys all cache entries living under oldPath so that +// they live under newPath instead. +func (c *IDCache) MovePath(ctx context.Context, oldPath, newPath string) error { + oldPath = filepath.Clean(oldPath) + newPath = filepath.Clean(newPath) + + // decode turns a KV record (reverse key + forward cache key value) into a + // moveEntry without issuing any additional Get calls. + decode := func(reverseKey string, value []byte) (moveEntry, error) { + spaceID, nodeID, err := decodeCacheKey(string(value)) + if err != nil { + return moveEntry{}, err + } + currentPath, err := pathFromReverseCacheKey(reverseKey) + if err != nil { + return moveEntry{}, err + } + return moveEntry{ + reverseKey: reverseKey, + spaceID: spaceID, + nodeID: nodeID, + currentPath: currentPath, + }, nil + } + + entries := make([]moveEntry, 0) + // the entry for the moved node itself + if record, err := retry(ctx, func() (jetstream.KeyValueEntry, error) { + return c.kv.Get(ctx, reverseCacheKey(oldPath)) + }); err == nil { + if e, derr := decode(record.Key(), record.Value()); derr == nil { + entries = append(entries, e) + } else { + appctx.GetLogger(ctx).Error().Err(derr).Str("record", record.Key()).Msg("could not decode cache entry") + } + } else if err != jetstream.ErrKeyNotFound { + return err + } + // all of its descendants + baseKey := reverseCacheKey(oldPath) + watcher, err := retry(ctx, func() (jetstream.KeyWatcher, error) { + return c.kv.Watch(ctx, baseKey+".>", jetstream.IgnoreDeletes()) + }) + if err != nil { + return err + } + for update := range watcher.Updates() { + if update == nil { + break + } + e, derr := decode(update.Key(), update.Value()) + if derr != nil { + appctx.GetLogger(ctx).Error().Err(derr).Str("record", update.Key()).Msg("could not decode cache entry") + continue + } + entries = append(entries, e) + } + _ = watcher.Stop() + + // Re-key all collected entries concurrently. Each entry results in two Puts + // (forward + reverse) and one Purge of the stale reverse key. + g, gctx := errgroup.WithContext(ctx) + g.SetLimit(movePathConcurrency) + for _, e := range entries { + g.Go(func() error { + // defensively make sure the entry really lives under oldPath + if e.currentPath != oldPath && !strings.HasPrefix(e.currentPath, oldPath+string(filepath.Separator)) { + return nil + } + + updatedPath := newPath + strings.TrimPrefix(e.currentPath, oldPath) + + // Set writes the new forward (id -> path) and reverse (path -> id) + // entries. The forward entry is overwritten in place, the now stale + // reverse entry for the old path is purged afterwards. + if err := c.Set(gctx, e.spaceID, e.nodeID, updatedPath); err != nil { + return err + } + return retryErr(gctx, func() error { + return c.kv.Purge(gctx, e.reverseKey) + }) + }) + } + return g.Wait() +} + // Set adds a new entry to the cache func (c *IDCache) Set(ctx context.Context, spaceID, nodeID, val string) error { _, err := retry(ctx, func() (uint64, error) { @@ -168,7 +266,13 @@ func (c *IDCache) getByReverseCacheKey(ctx context.Context, reverseKey string) ( } return "", "", err } - decoded, err := base32.StdEncoding.DecodeString(string(record.Value())) + return decodeCacheKey(string(record.Value())) +} + +// decodeCacheKey decodes an encoded forward cache key (base32 of +// "spaceID!nodeID") back into its spaceID and nodeID parts. +func decodeCacheKey(encoded string) (string, string, error) { + decoded, err := base32.StdEncoding.DecodeString(encoded) if err != nil { return "", "", err } @@ -198,6 +302,21 @@ func reverseCacheKey(path string) string { return strings.Join(encoded, ".") } +// pathFromReverseCacheKey reverses reverseCacheKey, decoding an encoded reverse +// cache key back into the absolute filesystem path it represents. +func pathFromReverseCacheKey(reverseKey string) (string, error) { + parts := strings.Split(reverseKey, ".") + decoded := make([]string, len(parts)) + for i, p := range parts { + b, err := base32.StdEncoding.DecodeString(p) + if err != nil { + return "", err + } + decoded[i] = string(b) + } + return string(filepath.Separator) + strings.Join(decoded, string(filepath.Separator)), nil +} + func retry[T any](ctx context.Context, f func() (T, error)) (T, error) { var v T var err error diff --git a/pkg/storage/fs/posix/idcache/idcache_test.go b/pkg/storage/fs/posix/idcache/idcache_test.go index 300e907b411..70ba65131d2 100644 --- a/pkg/storage/fs/posix/idcache/idcache_test.go +++ b/pkg/storage/fs/posix/idcache/idcache_test.go @@ -146,6 +146,53 @@ var _ = Describe("IDCache", func() { }) }) + Describe("MovePath", func() { + It("re-keys the moved node and its whole subtree", func() { + Expect(c.Set(context.TODO(), "spaceID", "nodeID", "/path")).To(Succeed()) + Expect(c.Set(context.TODO(), "spaceID", "nodeID2", "/path/child")).To(Succeed()) + Expect(c.Set(context.TODO(), "spaceID", "nodeID3", "/path/child/grandchild")).To(Succeed()) + + err := c.MovePath(context.TODO(), "/path", "/newpath") + Expect(err).ToNot(HaveOccurred()) + + // forward lookups now resolve to the new paths + v, err := c.Get(context.TODO(), "spaceID", "nodeID") + Expect(err).ToNot(HaveOccurred()) + Expect(v).To(Equal("/newpath")) + v, err = c.Get(context.TODO(), "spaceID", "nodeID2") + Expect(err).ToNot(HaveOccurred()) + Expect(v).To(Equal("/newpath/child")) + v, err = c.Get(context.TODO(), "spaceID", "nodeID3") + Expect(err).ToNot(HaveOccurred()) + Expect(v).To(Equal("/newpath/child/grandchild")) + + // reverse lookups resolve under the new path + spaceID, nodeID, err := c.GetByPath(context.TODO(), "/newpath") + Expect(err).ToNot(HaveOccurred()) + Expect(spaceID).To(Equal("spaceID")) + Expect(nodeID).To(Equal("nodeID")) + _, nodeID, err = c.GetByPath(context.TODO(), "/newpath/child") + Expect(err).ToNot(HaveOccurred()) + Expect(nodeID).To(Equal("nodeID2")) + _, nodeID, err = c.GetByPath(context.TODO(), "/newpath/child/grandchild") + Expect(err).ToNot(HaveOccurred()) + Expect(nodeID).To(Equal("nodeID3")) + + // the old reverse lookups are gone + _, _, err = c.GetByPath(context.TODO(), "/path") + Expect(err).To(HaveOccurred()) + _, _, err = c.GetByPath(context.TODO(), "/path/child") + Expect(err).To(HaveOccurred()) + _, _, err = c.GetByPath(context.TODO(), "/path/child/grandchild") + Expect(err).To(HaveOccurred()) + }) + + It("does not fail if the path does not exist", func() { + err := c.MovePath(context.TODO(), "/nonexistent", "/newpath") + Expect(err).ToNot(HaveOccurred()) + }) + }) + Describe("Retries", func() { It("should retry operations and succeed if transient errors resolve", func() { _, js, _, err := helpers.NewInProcessNATSServer() diff --git a/pkg/storage/fs/posix/lookup/lookup.go b/pkg/storage/fs/posix/lookup/lookup.go index 17c69678410..3b42aafcdf6 100644 --- a/pkg/storage/fs/posix/lookup/lookup.go +++ b/pkg/storage/fs/posix/lookup/lookup.go @@ -72,6 +72,8 @@ type IDCache interface { DeleteByPath(ctx context.Context, path string) error DeletePath(ctx context.Context, path string) error + + MovePath(ctx context.Context, oldPath, newPath string) error } // Lookup implements transformations from filepath to node and back diff --git a/pkg/storage/fs/posix/lookup/mocks/IDCache.go b/pkg/storage/fs/posix/lookup/mocks/IDCache.go index 3a4fabffcfd..1508b3398dd 100644 --- a/pkg/storage/fs/posix/lookup/mocks/IDCache.go +++ b/pkg/storage/fs/posix/lookup/mocks/IDCache.go @@ -181,6 +181,54 @@ func (_c *IDCache_DeletePath_Call) RunAndReturn(run func(context.Context, string return _c } +// MovePath provides a mock function with given fields: ctx, oldPath, newPath +func (_m *IDCache) MovePath(ctx context.Context, oldPath string, newPath string) error { + ret := _m.Called(ctx, oldPath, newPath) + + if len(ret) == 0 { + panic("no return value specified for MovePath") + } + + var r0 error + if rf, ok := ret.Get(0).(func(context.Context, string, string) error); ok { + r0 = rf(ctx, oldPath, newPath) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// IDCache_MovePath_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'MovePath' +type IDCache_MovePath_Call struct { + *mock.Call +} + +// MovePath is a helper method to define mock.On call +// - ctx context.Context +// - oldPath string +// - newPath string +func (_e *IDCache_Expecter) MovePath(ctx interface{}, oldPath interface{}, newPath interface{}) *IDCache_MovePath_Call { + return &IDCache_MovePath_Call{Call: _e.mock.On("MovePath", ctx, oldPath, newPath)} +} + +func (_c *IDCache_MovePath_Call) Run(run func(ctx context.Context, oldPath string, newPath string)) *IDCache_MovePath_Call { + _c.Call.Run(func(args mock.Arguments) { + run(args[0].(context.Context), args[1].(string), args[2].(string)) + }) + return _c +} + +func (_c *IDCache_MovePath_Call) Return(_a0 error) *IDCache_MovePath_Call { + _c.Call.Return(_a0) + return _c +} + +func (_c *IDCache_MovePath_Call) RunAndReturn(run func(context.Context, string, string) error) *IDCache_MovePath_Call { + _c.Call.Return(run) + return _c +} + // Get provides a mock function with given fields: ctx, spaceID, nodeID func (_m *IDCache) Get(ctx context.Context, spaceID string, nodeID string) (string, error) { ret := _m.Called(ctx, spaceID, nodeID) diff --git a/pkg/storage/fs/posix/tree/tree.go b/pkg/storage/fs/posix/tree/tree.go index 243417c4bab..b1d0e9b05da 100644 --- a/pkg/storage/fs/posix/tree/tree.go +++ b/pkg/storage/fs/posix/tree/tree.go @@ -466,14 +466,11 @@ func (t *Tree) Move(ctx context.Context, oldNode *node.Node, newNode *node.Node) subspan.End() _, subspan = tracer.Start(ctx, "update id cache and attributes") - // update the id cache - // invalidate old tree - err = t.lookup.IDCache.DeleteByPath(ctx, filepath.Join(oldNode.ParentPath(), oldNode.Name)) - if err != nil { - return err - } - if err := t.lookup.CacheID(ctx, newNode.SpaceID, newNode.ID, filepath.Join(newNode.ParentPath(), newNode.Name)); err != nil { - t.log.Error().Err(err).Str("spaceID", newNode.SpaceID).Str("id", newNode.ID).Str("path", filepath.Join(newNode.ParentPath(), newNode.Name)).Msg("could not cache id") + oldPath := filepath.Join(oldNode.ParentPath(), oldNode.Name) + newPath := filepath.Join(newNode.ParentPath(), newNode.Name) + + if err := t.lookup.CacheID(ctx, newNode.SpaceID, newNode.ID, newPath); err != nil { + t.log.Error().Err(err).Str("spaceID", newNode.SpaceID).Str("id", newNode.ID).Str("path", newPath).Msg("could not cache id") } // update target parentid and name @@ -520,15 +517,21 @@ func (t *Tree) Move(ctx context.Context, oldNode *node.Node, newNode *node.Node) } if oldNode.IsDir(ctx) { - go func() { - _, subspan = tracer.Start(ctx, "warmup id cache for moved subtree") - // update id cache for the moved subtree. - err = t.WarmupIDCache(filepath.Join(newNode.ParentPath(), newNode.Name), false, false) - if err != nil { - t.log.Error().Err(err).Str("path", filepath.Join(newNode.ParentPath(), newNode.Name)).Msg("failed to warmup id cache for moved subtree") - } - subspan.End() - }() + // Re-key the cached ids of the moved subtree instead of re-scanning it from disk. The node ids do not change on a move, + // only their paths do, so re-keying the existing cache entries is much cheaper than a full filesystem walk for large trees. + // MovePath also re-keys the moved node itself, which purges its now stale reverse (path -> id) cache entry for the old path. + start := time.Now() + if err := t.lookup.IDCache.MovePath(context.Background(), oldPath, newPath); err != nil { + t.log.Error().Err(err).Str("oldPath", oldPath).Str("newPath", newPath).Msg("failed to move id cache for moved subtree") + } + t.log.Info().Dur("duration", time.Since(start)).Str("oldPath", oldPath).Str("newPath", newPath).Msg("moved id cache for moved subtree") + } else { + // A file move has no subtree to re-key, but the now stale reverse (path -> id) cache entry for the old path must still be + // removed. Otherwise the old path keeps resolving to the moved node's id and, because InternalPath() recomputes the on-disk + // path from that id, the old path resolves to the moved (or a dangling) location instead of reporting the file as gone. + if err := t.lookup.IDCache.DeletePath(ctx, oldPath); err != nil { + t.log.Error().Err(err).Str("oldPath", oldPath).Msg("could not remove old path from id cache after move") + } } return nil diff --git a/pkg/storage/fs/posix/tree/tree_non_watching_test.go b/pkg/storage/fs/posix/tree/tree_non_watching_test.go index 29839c3e16b..dfe47037c52 100644 --- a/pkg/storage/fs/posix/tree/tree_non_watching_test.go +++ b/pkg/storage/fs/posix/tree/tree_non_watching_test.go @@ -106,6 +106,68 @@ var _ = Describe("Non-watching tree", func() { }).Should(Succeed()) }) + It("removes the old path from the id cache when moving a file", func() { + ctx := non_watching_env.Ctx + + // the subtree directory has been created in the BeforeEach + parent, err := non_watching_env.Lookup.NodeFromResource(ctx, &provider.Reference{ + ResourceId: non_watching_env.SpaceRootRes, + Path: subtree, + }) + Expect(err).ToNot(HaveOccurred()) + Expect(parent.Exists).To(BeTrue()) + + // create the target subdirectory to move the file into + targetDir, err := non_watching_env.Lookup.NodeFromResource(ctx, &provider.Reference{ + ResourceId: non_watching_env.SpaceRootRes, + Path: subtree + "/subdir", + }) + Expect(err).ToNot(HaveOccurred()) + Expect(non_watching_env.Tree.CreateDir(ctx, targetDir)).To(Succeed()) + + // create the source file directly below the subtree + _, err = non_watching_env.CreateTestFile("source.txt", "source-blob", parent.ID, parent.SpaceID, 1) + Expect(err).ToNot(HaveOccurred()) + + oldPath := filepath.Join(root, "source.txt") + + // sanity check: the old path is known to the id cache + _, _, err = non_watching_env.Lookup.IDCache.GetByPath(ctx, oldPath) + Expect(err).ToNot(HaveOccurred()) + + // resolve source and target nodes just like the fs Move does + oldNode, err := non_watching_env.Lookup.NodeFromResource(ctx, &provider.Reference{ + ResourceId: non_watching_env.SpaceRootRes, + Path: subtree + "/source.txt", + }) + Expect(err).ToNot(HaveOccurred()) + Expect(oldNode.Exists).To(BeTrue()) + + newNode, err := non_watching_env.Lookup.NodeFromResource(ctx, &provider.Reference{ + ResourceId: non_watching_env.SpaceRootRes, + Path: subtree + "/subdir/moved.txt", + }) + Expect(err).ToNot(HaveOccurred()) + Expect(newNode.Exists).To(BeFalse()) + + // move the file + Expect(non_watching_env.Tree.Move(ctx, oldNode, newNode)).To(Succeed()) + + // the stale reverse (path -> id) cache entry for the old path must be purged + _, _, err = non_watching_env.Lookup.IDCache.GetByPath(ctx, oldPath) + Expect(err).To(HaveOccurred()) + _, ok := err.(errtypes.NotFound) + Expect(ok).To(BeTrue(), "old path should no longer resolve in the id cache after a move") + + // resolving the old path must report the file as gone, not the moved node + stale, err := non_watching_env.Lookup.NodeFromResource(ctx, &provider.Reference{ + ResourceId: non_watching_env.SpaceRootRes, + Path: subtree + "/source.txt", + }) + Expect(err).ToNot(HaveOccurred()) + Expect(stale.Exists).To(BeFalse(), "old path must not resolve to the moved node") + }) + It("rejects creation of internal paths", func() { spaceRoot := non_watching_env.Root + "/users/" + non_watching_env.Owner.Username