Re-key cache entries of moved items instead of scanning from disk again

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

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()

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

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

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