diff --git a/weed/server/filer_grpc_server_traverse_meta.go b/weed/server/filer_grpc_server_traverse_meta.go index 9e317e83f..2ebf33d05 100644 --- a/weed/server/filer_grpc_server_traverse_meta.go +++ b/weed/server/filer_grpc_server_traverse_meta.go @@ -22,34 +22,63 @@ func (fs *FilerServer) TraverseBfsMetadata(req *filer_pb.TraverseBfsMetadataRequ ctx := stream.Context() - queue := util.NewQueue[*filer.Entry]() + isExcluded := func(path util.FullPath) bool { + excluded := false + excludedTrie.MatchPrefix([]byte(path), func(key []byte, value bool) bool { + // Only exclude when the prefix ends on a path-component boundary, + // so /a/b does not also exclude a sibling like /a/bc. + if len(path) == len(key) || path[len(key)] == '/' { + excluded = true + return false + } + return true + }) + return excluded + } + + // Send each entry as it is visited and queue only directory paths to + // descend into. Queuing the entries themselves would hold the whole + // subtree in memory, including every file's chunk list, and can exhaust + // the filer on large trees (e.g. during a peer's first-time bootstrap). + sendEntry := func(entry *filer.Entry) error { + parent, _ := entry.FullPath.DirAndName() + if err := stream.Send(&filer_pb.TraverseBfsMetadataResponse{ + Directory: parent, + Entry: entry.ToProtoEntry(), + }); err != nil { + return fmt.Errorf("send traverse bfs metadata response: %w", err) + } + return nil + } + dirEntry, err := fs.filer.FindEntry(ctx, util.FullPath(req.Directory)) if err != nil { return fmt.Errorf("find dir %s: %v", req.Directory, err) } - queue.Enqueue(dirEntry) + if isExcluded(dirEntry.FullPath) { + return nil + } + if err := sendEntry(dirEntry); err != nil { + return err + } - for item := queue.Dequeue(); item != nil; item = queue.Dequeue() { - if excludedTrie.MatchPrefix([]byte(item.FullPath), func(key []byte, value bool) bool { - return true - }) { - // println("excluded", item.FullPath) - continue - } - parent, _ := item.FullPath.DirAndName() - if err := stream.Send(&filer_pb.TraverseBfsMetadataResponse{ - Directory: parent, - Entry: item.ToProtoEntry(), - }); err != nil { - return fmt.Errorf("send traverse bfs metadata response: %w", err) - } + queue := util.NewQueue[util.FullPath]() + if dirEntry.IsDirectory() { + queue.Enqueue(dirEntry.FullPath) + } - if !item.IsDirectory() { - continue - } - - if err := fs.iterateDirectory(ctx, item.FullPath, func(entry *filer.Entry) error { - queue.Enqueue(entry) + for queue.Len() > 0 { + dirPath := queue.Dequeue() + if err := fs.iterateDirectory(ctx, dirPath, func(entry *filer.Entry) error { + if isExcluded(entry.FullPath) { + return nil + } + if err := sendEntry(entry); err != nil { + return err + } + if entry.IsDirectory() { + queue.Enqueue(entry.FullPath) + } return nil }); err != nil { return err diff --git a/weed/server/filer_grpc_server_traverse_meta_test.go b/weed/server/filer_grpc_server_traverse_meta_test.go index 31cd33064..7aa9e3a2e 100644 --- a/weed/server/filer_grpc_server_traverse_meta_test.go +++ b/weed/server/filer_grpc_server_traverse_meta_test.go @@ -1,10 +1,14 @@ package weed_server import ( + "context" "testing" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" "github.com/stretchr/testify/assert" "github.com/viant/ptrie" + "google.golang.org/grpc" ) func TestPtrie(t *testing.T) { @@ -30,3 +34,61 @@ func TestPtrie(t *testing.T) { assert.False(t, excludedTrie.Has(b)) } + +type captureTraverseStream struct { + grpc.ServerStream + ctx context.Context + visited []string +} + +func (s *captureTraverseStream) Context() context.Context { return s.ctx } + +func (s *captureTraverseStream) Send(resp *filer_pb.TraverseBfsMetadataResponse) error { + dir := resp.Directory + if dir != "/" { + dir += "/" + } + s.visited = append(s.visited, dir+resp.Entry.Name) + return nil +} + +func TestTraverseBfsMetadata(t *testing.T) { + store := newRenameTestStore() + // "/.system-data" shares the "/.system" byte prefix but is a different + // path component, so it must not be excluded. + for _, dir := range []string{"/", "/a", "/a/sub", "/b", "/.system", "/.system-data"} { + store.entries[dir] = newDirectoryEntry(dir, 1) + } + for _, file := range []string{"/a/f1", "/a/sub/f2", "/b/f3", "/.system/secret", "/.system-data/keep"} { + store.entries[file] = newFileEntry(file, 2) + } + + server := &FilerServer{filer: newRenameTestFiler(store)} + stream := &captureTraverseStream{ctx: context.Background()} + + err := server.TraverseBfsMetadata(&filer_pb.TraverseBfsMetadataRequest{ + Directory: "/", + ExcludedPrefixes: []string{"/.system"}, + }, stream) + assert.NoError(t, err) + + // The excluded subtree is skipped; everything else (including the + // /.system-data sibling) is visited exactly once. + assert.ElementsMatch(t, []string{ + "/", "/a", "/a/f1", "/a/sub", "/a/sub/f2", "/b", "/b/f3", + "/.system-data", "/.system-data/keep", + }, stream.visited) + + // Each entry's parent must be streamed before the entry itself, so a + // consumer can apply the tree top-down. + seen := make(map[string]bool) + for _, path := range stream.visited { + if path != "/" { + parent, _ := util.FullPath(path).DirAndName() + assert.Truef(t, seen[parent], "parent %s of %s not streamed first", parent, path) + } + seen[path] = true + } +} + +var _ filer_pb.SeaweedFiler_TraverseBfsMetadataServer = (*captureTraverseStream)(nil)