diff --git a/weed/pb/filer_pb/filer_client.go b/weed/pb/filer_pb/filer_client.go index aa2425885..1fd86de7e 100644 --- a/weed/pb/filer_pb/filer_client.go +++ b/weed/pb/filer_pb/filer_client.go @@ -91,9 +91,15 @@ func ReadDirAllEntriesWithSnapshot(ctx context.Context, filerClient FilerClient, for counter == paginationLimit { counter = 0 + lastStartFrom := startFrom if _, err = doListWithSnapshot(ctx, filerClient, fullDirPath, prefix, counterFunc, startFrom, false, paginationLimit, snapshotTsNs); err != nil { return snapshotTsNs, err } + // A full page that ends on the name it started from would loop forever; + // a store whose ordering does not advance past the cursor causes this. + if counter == paginationLimit && startFrom == lastStartFrom { + return snapshotTsNs, fmt.Errorf("list %s: pagination stuck at %q", fullDirPath, startFrom) + } } return snapshotTsNs, nil diff --git a/weed/pb/filer_pb/filer_client_bfs.go b/weed/pb/filer_pb/filer_client_bfs.go index f6dcdf880..451b58511 100644 --- a/weed/pb/filer_pb/filer_client_bfs.go +++ b/weed/pb/filer_pb/filer_client_bfs.go @@ -26,11 +26,21 @@ func TraverseBfs(ctx context.Context, filerClient FilerClient, parentPath util.F var once sync.Once var firstErr error + // A directory delivered twice (a page-boundary race with concurrent + // renames, or a store whose listing order misbehaves) must not be walked + // twice: the second walk re-lists the same subtree and can keep the + // traversal from ever terminating. + var visited sync.Map + visited.Store(string(parentPath), struct{}{}) + enqueue := func(p util.FullPath) bool { // Stop expanding traversal once canceled (e.g. first error encountered). if ctx.Err() != nil { return false } + if _, seen := visited.LoadOrStore(string(p), struct{}{}); seen { + return true + } pending.Add(1) queue.Enqueue(p) return true diff --git a/weed/pb/filer_pb/filer_client_bfs_test.go b/weed/pb/filer_pb/filer_client_bfs_test.go new file mode 100644 index 000000000..c9ec79e14 --- /dev/null +++ b/weed/pb/filer_pb/filer_client_bfs_test.go @@ -0,0 +1,152 @@ +package filer_pb + +import ( + "context" + "fmt" + "net" + "strings" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// stubFiler serves canned listings and counts how often each directory is listed. +type stubFiler struct { + UnimplementedSeaweedFilerServer + mu sync.Mutex + listed map[string]int + listings func(req *ListEntriesRequest, send func(*Entry) error) error +} + +func (s *stubFiler) ListEntries(req *ListEntriesRequest, stream SeaweedFiler_ListEntriesServer) error { + s.mu.Lock() + if s.listed == nil { + s.listed = make(map[string]int) + } + s.listed[req.Directory]++ + s.mu.Unlock() + return s.listings(req, func(entry *Entry) error { + return stream.Send(&ListEntriesResponse{Entry: entry}) + }) +} + +func (s *stubFiler) listedCount(dir string) int { + s.mu.Lock() + defer s.mu.Unlock() + return s.listed[dir] +} + +type stubFilerClient struct { + conn *grpc.ClientConn +} + +func (c *stubFilerClient) WithFilerClient(streamingMode bool, fn func(SeaweedFilerClient) error) error { + return fn(NewSeaweedFilerClient(c.conn)) +} + +func (c *stubFilerClient) AdjustedUrl(location *Location) string { return location.Url } + +func (c *stubFilerClient) GetDataCenter() string { return "" } + +func startStubFiler(t *testing.T, s *stubFiler) *stubFilerClient { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + grpcServer := grpc.NewServer() + RegisterSeaweedFilerServer(grpcServer, s) + go grpcServer.Serve(listener) + t.Cleanup(grpcServer.Stop) + conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.Close() }) + return &stubFilerClient{conn: conn} +} + +// A directory delivered twice by a misbehaving listing must still be walked once. +func TestTraverseBfsVisitsDuplicateDirectoryOnce(t *testing.T) { + server := &stubFiler{} + server.listings = func(req *ListEntriesRequest, send func(*Entry) error) error { + switch req.Directory { + case "/": + if req.StartFromFileName != "" { + return nil + } + for _, name := range []string{"dup", "dup"} { + if err := send(&Entry{Name: name, IsDirectory: true}); err != nil { + return err + } + } + case "/dup": + if req.StartFromFileName != "" { + return nil + } + for _, name := range []string{"a.txt", "b.txt"} { + if err := send(&Entry{Name: name}); err != nil { + return err + } + } + } + return nil + } + filerClient := startStubFiler(t, server) + + var mu sync.Mutex + seen := make(map[string]int) + err := TraverseBfs(context.Background(), filerClient, "/", func(parentPath util.FullPath, entry *Entry) error { + mu.Lock() + seen[string(parentPath.Child(entry.Name))]++ + mu.Unlock() + return nil + }) + if err != nil { + t.Fatal(err) + } + + if got := server.listedCount("/dup"); got != 1 { + t.Fatalf("listed /dup %d times, want 1", got) + } + for _, path := range []string{"/dup/a.txt", "/dup/b.txt"} { + if seen[path] != 1 { + t.Fatalf("visited %s %d times, want 1", path, seen[path]) + } + } +} + +// A store whose pagination never advances past the cursor must error out +// instead of re-listing the same page forever. +func TestReadDirAllEntriesStuckPagination(t *testing.T) { + server := &stubFiler{} + server.listings = func(req *ListEntriesRequest, send func(*Entry) error) error { + for i := uint32(0); i < req.Limit; i++ { + if err := send(&Entry{Name: fmt.Sprintf("f%05d", i)}); err != nil { + return err + } + } + return nil + } + filerClient := startStubFiler(t, server) + + done := make(chan error, 1) + go func() { + done <- ReadDirAllEntries(context.Background(), filerClient, "/", "", func(entry *Entry, isLast bool) error { + return nil + }) + }() + + select { + case err := <-done: + if err == nil || !strings.Contains(err.Error(), "pagination stuck") { + t.Fatalf("want pagination stuck error, got %v", err) + } + case <-time.After(30 * time.Second): + t.Fatal("listing loops forever on a non-advancing store") + } +} diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go index 2156c6f3a..cefda2634 100644 --- a/weed/shell/command_fs_merge_volumes.go +++ b/weed/shell/command_fs_merge_volumes.go @@ -139,6 +139,13 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer // synchronize without a global lock. var processedHardLinks sync.Map + planCollections := make(map[string]bool) + for src := range plan.targets { + if info := c.volumes[src]; info != nil { + planCollections[info.Collection] = true + } + } + return commandEnv.WithFilerClient(false, func(filerClient filer_pb.SeaweedFilerClient) error { return filer_pb.TraverseBfs(context.Background(), commandEnv, util.FullPath(dir), func(parentPath util.FullPath, entry *filer_pb.Entry) error { if entry.IsDirectory { @@ -164,6 +171,9 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer var movedSources []movedSourceNeedle for i, chunk := range entry.Chunks { if chunk.IsChunkManifest { + if !c.manifestMayReferencePlan(plan, planCollections, needle.VolumeId(chunk.Fid.VolumeId)) { + continue + } oldManifestFid := chunk.GetFileIdString() oldManifestVid := chunk.Fid.VolumeId newChunk, changed, subSources, mErr := c.rewriteManifestChunk(context.Background(), commandEnv, lookupFn, plan, entryPath, chunk, *apply) @@ -292,6 +302,18 @@ func (c *commandFsMergeVolumes) deleteMovedSourceNeedles(commandEnv *CommandEnv, } } +// Sub-chunks are assigned in the manifest's own collection, so a manifest on +// a volume in a collection the plan does not touch cannot reference any plan +// volume — resolving it would just download manifest needles across the whole +// namespace for nothing. +func (c *commandFsMergeVolumes) manifestMayReferencePlan(plan *mergePlan, planCollections map[string]bool, vid needle.VolumeId) bool { + if plan.isSource(vid) { + return true + } + info := c.volumes[vid] + return info == nil || planCollections[info.Collection] +} + func (c *commandFsMergeVolumes) getVolumeInfoById(vid needle.VolumeId) (*master_pb.VolumeInformationMessage, error) { info := c.volumes[vid] var err error diff --git a/weed/shell/command_fs_merge_volumes_test.go b/weed/shell/command_fs_merge_volumes_test.go index 859ba80b4..3582692c7 100644 --- a/weed/shell/command_fs_merge_volumes_test.go +++ b/weed/shell/command_fs_merge_volumes_test.go @@ -246,3 +246,28 @@ func TestGetVolumeSize_ClampsDeletedOverSize(t *testing.T) { t.Errorf("expected 2000, got %d", got) } } + +// Manifests on volumes in a foreign collection cannot reference plan volumes +// and must be skipped instead of downloaded; plan sources and unknown volumes +// must still be resolved. +func TestManifestMayReferencePlan(t *testing.T) { + src := &master_pb.VolumeInformationMessage{Id: 1, Size: 100, Collection: "x"} + same := &master_pb.VolumeInformationMessage{Id: 2, Size: 100, Collection: "x"} + foreign := &master_pb.VolumeInformationMessage{Id: 3, Size: 100, Collection: "y"} + c := newMergeCmd(250000, src, same, foreign) + + plan := newMergePlan(c.volumeSizeLimit) + plan.targets[needle.VolumeId(1)] = []needle.VolumeId{2} + planCollections := map[string]bool{"x": true} + + for vid, want := range map[needle.VolumeId]bool{ + 1: true, // plan source + 2: true, // same collection + 3: false, // foreign collection + 9: true, // unknown volume, resolve conservatively + } { + if got := c.manifestMayReferencePlan(plan, planCollections, vid); got != want { + t.Fatalf("volume %d: got %v, want %v", vid, got, want) + } + } +}