filer: bound TraverseBfsMetadata memory by queuing directory paths (#9814)

* filer: bound TraverseBfsMetadata memory by queuing directory paths

The BFS enqueued every entry, so it held the whole subtree in memory
including each file's chunk list. A filer serving a peer's first-time
bootstrap traversal of a large tree could exhaust memory and get killed.

Stream each entry as it is visited and queue only directory paths to
descend into. Memory is now bounded by the number of directories rather
than the entire tree, and the streamed output order is unchanged.

* filer: match excluded prefixes on path-component boundaries

Only treat an excluded prefix as a match when it ends at a path
boundary, so excluding /a/b does not also drop a sibling like /a/bc.
Short-circuit the trie walk on the first real match.

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

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)