From 641ab6204b2fe040a7b63bb6840f769f09348cef Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 11 Nov 2026 09:55:03 -0800 Subject: [PATCH] shell: fs.mergeVolumes leaves in-progress multipart parts in place (#12556) * shell: fs.mergeVolumes leaves in-progress multipart parts in place CompleteMultipartUpload reads the part entries and later commits the object with their fids, guarded only by the upload directory. A merge that moved a part chunk in between deleted the source needle the new object pointed at. Parts stay where they are until the upload completes or is cleaned, and their chunks still count as referenced sources. Co-Authored-By: Claude Opus 5.5 * shell: note in fs.mergeVolumes help that upload parts stay in place Co-Authored-By: Claude Opus 5.5 * shell: match upload parts when the buckets folder is the root Co-Authored-By: Claude Opus 5.5 * shell: count chunks behind a part's manifest as referenced sources A skipped part whose chunks on a merge source sit behind a manifest was not counted, so the merge printed no skip line and warned that nothing references the source. Co-Authored-By: Claude Opus 5.5 * shell: keep a part's direct source references when its manifest is unreadable Co-Authored-By: Claude Opus 5.5 * shell: count chunks resolved before an unreadable part manifest ResolveChunkManifest returns the data chunks of earlier manifests along with a later manifest's error; record them before returning so the unreferenced-source warning does not fire for a source they reference. Co-Authored-By: Claude Opus 5.5 * shell: test the partial manifest result without a timing delay The unreadable manifest is now nested under a readable part manifest, so the first level resolves fully before the failing lookup runs. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/shell/command_fs_merge_volumes.go | 64 ++++++++++ weed/shell/command_fs_merge_volumes_test.go | 122 ++++++++++++++++++++ 2 files changed, 186 insertions(+) diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go index 26059da18e..5bf8c49d26 100644 --- a/weed/shell/command_fs_merge_volumes.go +++ b/weed/shell/command_fs_merge_volumes.go @@ -8,6 +8,7 @@ import ( "io" "math" "net/http" + "path" "sort" "strconv" "strings" @@ -17,6 +18,7 @@ import ( "slices" "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/util/wildcard" @@ -60,6 +62,9 @@ func (c *commandFsMergeVolumes) Help() string { -apply needs the shell lock (run "lock" first), so two runs never move the same chunks at once. + + Parts of in-progress S3 multipart uploads are not moved, so a source volume + holding them is not cleared until those uploads complete or are cleaned up. ` } @@ -160,12 +165,29 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer seen := newSourceNeedleCounter() + bucketsPath, err := readFilerBucketsPath(commandEnv) + if err != nil { + return err + } + if err := 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 { return nil } entryPath := parentPath.Child(entry.Name) + // A completion may already have read this part's fids and will + // commit the object with them, so moving the chunk would leave the + // object pointing at the deleted source needle. + if isMultipartUploadFile(bucketsPath, parentPath) { + found, rErr := c.recordPlanSources(context.Background(), lookupFn, plan, planCollections, seen, entry.Chunks) + if rErr != nil { + fmt.Printf("skip %s: part of an in-progress multipart upload, manifest unreadable: %v\n", entryPath, rErr) + } else if found { + fmt.Printf("skip %s: part of an in-progress multipart upload\n", entryPath) + } + return nil + } if len(entry.HardLinkId) > 0 { if _, seen := processedHardLinks.LoadOrStore(string(entry.HardLinkId), struct{}{}); seen { // Another sibling already carried the HardLinkId through @@ -283,6 +305,48 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer return nil } +// isMultipartUploadFile reports whether entries in dir are the parts of an +// in-progress multipart upload, //.uploads/. +func isMultipartUploadFile(bucketsPath string, dir util.FullPath) bool { + uploadsDir, _ := dir.DirAndName() + bucketDir, uploads := util.FullPath(uploadsDir).DirAndName() + if uploads != s3_constants.MultipartUploadsFolder { + return false + } + parent, _ := util.FullPath(bucketDir).DirAndName() + return parent == path.Clean(bucketsPath) +} + +// recordPlanSources counts the chunks on plan sources as referenced, including +// those behind a manifest that may reference the plan, and reports whether +// there were any. References read before an unreadable manifest stay counted. +func (c *commandFsMergeVolumes) recordPlanSources(ctx context.Context, lookupFn wdclient.LookupFileIdFunctionType, plan *mergePlan, planCollections map[string]bool, seen *sourceNeedleCounter, chunks []*filer_pb.FileChunk) (bool, error) { + found := false + record := func(chunk *filer_pb.FileChunk) { + if vid := needle.VolumeId(chunk.Fid.GetVolumeId()); plan.isSource(vid) { + seen.record(vid) + found = true + } + } + var manifests []*filer_pb.FileChunk + for _, chunk := range chunks { + record(chunk) + if chunk.IsChunkManifest && c.manifestMayReferencePlan(plan, planCollections, needle.VolumeId(chunk.Fid.GetVolumeId())) { + manifests = append(manifests, chunk) + } + } + if len(manifests) == 0 { + return found, nil + } + dataChunks, manifestChunks, err := filer.ResolveChunkManifest(ctx, lookupFn, manifests, 0, math.MaxInt64, nil) + for _, chunk := range slices.Concat(dataChunks, manifestChunks) { + if !slices.Contains(manifests, chunk) { + record(chunk) + } + } + return found, err +} + // warnUnreferencedSources warns when a plan source's index still holds needles // but no filer entry referenced any during traversal — the merge moves nothing // and the real cleanup is volume.fsck. diff --git a/weed/shell/command_fs_merge_volumes_test.go b/weed/shell/command_fs_merge_volumes_test.go index 37500e2d45..17c71986e0 100644 --- a/weed/shell/command_fs_merge_volumes_test.go +++ b/weed/shell/command_fs_merge_volumes_test.go @@ -13,6 +13,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -513,3 +514,124 @@ func TestUnreferencedTargets(t *testing.T) { }) } } + +func TestIsMultipartUploadFile(t *testing.T) { + cases := []struct { + buckets, dir string + want bool + }{ + {"/buckets", "/buckets/b/.uploads/abc", true}, + {"/", "/b/.uploads/abc", true}, + {"/", "/buckets/b/.uploads/abc", false}, + {"/buckets/", "/buckets/b/.uploads/abc", true}, + {"/data/buckets", "/data/buckets/b/.uploads/abc", true}, + {"/buckets", "/buckets/b/.uploads", false}, + {"/buckets", "/buckets/b/dir/.uploads/abc", false}, + {"/buckets", "/buckets/b/.uploads/abc/sub", false}, + {"/buckets", "/other/b/.uploads/abc", false}, + {"/buckets", "/buckets/b/obj", false}, + } + for _, tc := range cases { + if got := isMultipartUploadFile(tc.buckets, util.FullPath(tc.dir)); got != tc.want { + t.Errorf("isMultipartUploadFile(%q, %q) = %v, want %v", tc.buckets, tc.dir, got, tc.want) + } + } +} + +func TestRecordPlanSources(t *testing.T) { + plan := &mergePlan{targets: map[needle.VolumeId][]needle.VolumeId{ + needle.VolumeId(203): {needle.VolumeId(187)}, + }} + c := newMergeCmd(250000, + &master_pb.VolumeInformationMessage{Id: 203, Collection: "b"}, + &master_pb.VolumeInformationMessage{Id: 187, Collection: "b"}, + &master_pb.VolumeInformationMessage{Id: 5, Collection: "other"}) + planCollections := map[string]bool{"b": true} + chunk := func(vid uint32, key uint64) *filer_pb.FileChunk { + fid := &filer_pb.FileId{VolumeId: vid, FileKey: key, Cookie: 7} + return &filer_pb.FileChunk{Fid: fid, FileId: needle.NewFileId(needle.VolumeId(vid), key, 7).String(), Size: 10} + } + + blob, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: []*filer_pb.FileChunk{chunk(203, 2)}}) + if err != nil { + t.Fatal(err) + } + volumeServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write(blob) + })) + defer volumeServer.Close() + lookupFn := func(_ context.Context, fileId string) ([]string, error) { + return []string{volumeServer.URL + "/" + fileId}, nil + } + manifest := func(vid uint32) *filer_pb.FileChunk { + m := chunk(vid, 1) + m.IsChunkManifest = true + return m + } + + for _, tc := range []struct { + name string + chunks []*filer_pb.FileChunk + found bool + count int + }{ + {"off plan sources", []*filer_pb.FileChunk{chunk(187, 1), chunk(5, 1)}, false, 0}, + {"on a plan source", []*filer_pb.FileChunk{chunk(203, 1), chunk(187, 2), chunk(203, 3)}, true, 2}, + {"behind a manifest off the source", []*filer_pb.FileChunk{manifest(187)}, true, 1}, + {"manifest outside the plan collections", []*filer_pb.FileChunk{manifest(5)}, false, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + seen := newSourceNeedleCounter() + found, err := c.recordPlanSources(context.Background(), lookupFn, plan, planCollections, seen, tc.chunks) + if err != nil { + t.Fatal(err) + } + if found != tc.found || seen.count(needle.VolumeId(203)) != tc.count { + t.Fatalf("found %v with %d references to 203, want %v with %d", found, seen.count(needle.VolumeId(203)), tc.found, tc.count) + } + }) + } + + unreachable := func(_ context.Context, fileId string) ([]string, error) { + return nil, fmt.Errorf("volume of %s unavailable", fileId) + } + seen := newSourceNeedleCounter() + found, err := c.recordPlanSources(context.Background(), unreachable, plan, planCollections, seen, []*filer_pb.FileChunk{chunk(203, 1), manifest(187)}) + if err == nil { + t.Fatal("an unreadable manifest must be reported") + } + if !found || seen.count(needle.VolumeId(203)) != 1 { + t.Fatalf("direct reference lost with an unreadable manifest: found %v, %d references to 203", found, seen.count(needle.VolumeId(203))) + } + + unreadable := chunk(187, 10) + unreadable.IsChunkManifest = true + nestedBlob, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: []*filer_pb.FileChunk{unreadable}}) + if err != nil { + t.Fatal(err) + } + outer := chunk(187, 9) + outer.IsChunkManifest = true + partServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/"+outer.GetFileIdString() { + _, _ = w.Write(nestedBlob) + return + } + _, _ = w.Write(blob) + })) + defer partServer.Close() + failsLater := func(_ context.Context, fileId string) ([]string, error) { + if fileId == unreadable.GetFileIdString() { + return nil, fmt.Errorf("volume of %s unavailable", fileId) + } + return []string{partServer.URL + "/" + fileId}, nil + } + seen = newSourceNeedleCounter() + found, err = c.recordPlanSources(context.Background(), failsLater, plan, planCollections, seen, []*filer_pb.FileChunk{manifest(187), outer}) + if err == nil { + t.Fatal("an unreadable manifest must be reported") + } + if !found || seen.count(needle.VolumeId(203)) != 1 { + t.Fatalf("chunks behind a readable manifest lost when a later one is unreadable: found %v, %d references to 203", found, seen.count(needle.VolumeId(203))) + } +}