From d13d975b6e66c01f4d00d9f8cfb82339c52d9daa Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 3 Nov 2026 07:59:18 -0800 Subject: [PATCH] shell: keep fs.mergeVolumes and fs.distributeChunks from reverting entries written while they run (#12362) * shell: keep fs.mergeVolumes from reverting entries written during the merge The walk reads a page of entries and moves their chunks one at a time, then wrote each entry back unconditionally. A file overwritten in between got its old version back, and the filer deleted the new version's chunks as garbage, so an acknowledged write was lost without any error. The update now carries IF_ENTRY_EQUAL over the entry as the walk read it. When the filer refuses because the entry changed or was deleted, the copies made on the target volumes are deleted, their plan room is released, and the entry is skipped; the source needles stay where the filer still points. Co-Authored-By: Claude Opus 5.5 * shell: keep fs.distributeChunks from reverting a file overwritten while it runs The command looks the entry up once, moves its chunks, and then wrote that entry back unconditionally, so an overwrite during the redistribution was reverted and the filer deleted the new version's chunks as garbage. The update now carries IF_ENTRY_EQUAL over the looked-up entry. When the filer refuses because the file changed or was deleted, the moved copies and any new manifest chunks are deleted and the command reports it. Co-Authored-By: Claude Opus 5.5 * shell: keep s3.versions.audit -heal from clearing a pointer that just moved The heal cleared the latest-version pointer on the directory entry the walk read, unconditionally. A version written between the audit and the heal has already pointed the directory at a live file, and clearing that pointer hid it. The heal now carries IF_ENTRY_EQUAL over the audited entry and reports a refusal as a failed heal. Co-Authored-By: Claude Opus 5.5 * shell: keep fs.meta.changeVolumeId from reverting entries written meanwhile The walk rewrote each entry's volume ids and wrote back the copy it had read, unconditionally, so an overwrite between the read and the write was reverted and its chunks were deleted as garbage. The update now carries IF_ENTRY_EQUAL over the entry as read, and a refused entry is skipped and reported. Co-Authored-By: Claude Opus 5.5 * shell: keep the target copies another fs.mergeVolumes run committed moveChunk keeps the key and cookie, so two overlapping merges of the same source into the same target write the same fid. When one run commits, the other's update is refused, and deleting its copies deleted the needle the committed entry now points at. A refused update now re-reads the entry, resolving its manifests, and deletes only the copies it no longer references. If the entry cannot be read, the copies are left for volume.fsck. Co-Authored-By: Claude Opus 5.5 * shell: let a later hard-link sibling retry a merge the filer refused The first sibling of a HardLinkId claims the move for all of them. When its update is refused, the claim made every later sibling skip, so the shared chunks stayed on the source volume for the rest of the run. A refused update now releases the claim, and a sibling listed after the change retries the move against the entry as it is now. Co-Authored-By: Claude Opus 5.5 * shell: delete the manifest blobs a failed fold in fs.distributeChunks saved When re-manifestizing failed partway, the blobs already uploaded were only counted in a log line and stayed on their volumes with nothing referencing them. They have freshly assigned fids, so they are now deleted with the cookie-checked delete the other orphan cleanups use. Co-Authored-By: Claude Opus 5.5 * shell: check the heal clears the pointer file name too Co-Authored-By: Claude Opus 5.5 * shell: require the shell lock for fs.mergeVolumes -apply moveChunk keeps the key and cookie, so two overlapping runs that move the same chunk into the same volume write the same fid. Re-reading the entry before deleting a refused run's copies covers the run that already committed, but not one that commits after that read. With the lock, two applying runs can no longer overlap, as volume.fsck and volume.merge already require. A dry run still works without it. Co-Authored-By: Claude Opus 5.5 * shell: check unreferencedTargets reads the refused entry itself Co-Authored-By: Claude Opus 5.5 * shell: check unreferencedTargets keeps a copy a manifest references Co-Authored-By: Claude Opus 5.5 * shell: check unreferencedTargets keeps the manifest copy and skips an unreadable manifest Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/shell/command_fs_distribute_chunks.go | 37 +++++- weed/shell/command_fs_merge_volumes.go | 119 +++++++++++++++--- weed/shell/command_fs_merge_volumes_test.go | 119 ++++++++++++++++++ .../shell/command_fs_meta_change_volume_id.go | 12 +- weed/shell/command_s3_versions_audit.go | 9 +- weed/shell/command_s3_versions_audit_test.go | 37 ++++++ 6 files changed, 300 insertions(+), 33 deletions(-) create mode 100644 weed/shell/command_s3_versions_audit_test.go diff --git a/weed/shell/command_fs_distribute_chunks.go b/weed/shell/command_fs_distribute_chunks.go index 17c6f4fdd9..bb05ce6217 100644 --- a/weed/shell/command_fs_distribute_chunks.go +++ b/weed/shell/command_fs_distribute_chunks.go @@ -23,6 +23,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/util" util_http "github.com/seaweedfs/seaweedfs/weed/util/http" + "google.golang.org/protobuf/proto" ) func init() { @@ -116,6 +117,7 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr if err != nil { return fmt.Errorf("lookup %s: %v", path, err) } + readEntry := proto.Clone(entry).(*filer_pb.Entry) chunks, hadManifest, err := resolveEntryDataChunks(commandEnv, entry) if err != nil { @@ -180,6 +182,11 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr fmt.Fprintf(writer, "\nExecuting redistribution...\n") defer util_http.GetGlobalHttpClient().CloseIdleConnections() + readFids := make(map[string]bool, len(chunks)) + for _, chunk := range chunks { + readFids[chunk.GetFileIdString()] = true + } + movedCount, err := executeChunkMoves(commandEnv, writer, moves, volumeInfoMap) if err != nil { return err @@ -191,8 +198,14 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr finalChunks := chunks if hadManifest { - // no chunk deleter here: a failed fold reports the blobs it saved - remanifested, manErr := filer.MaybeManifestize(newShellSaveAsChunk(commandEnv), nil, chunks) + dropSaved := func(saved []*filer_pb.FileChunk) { + var blobs []orphanedNeedle + for _, chunk := range saved { + blobs = append(blobs, orphanedNeedle{volumeId: chunk.GetFid().GetVolumeId(), fileId: chunk.GetFileIdString()}) + } + deleteOrphanedNeedles(commandEnv, util.FullPath(path), blobs, true) + } + remanifested, manErr := filer.MaybeManifestize(newShellSaveAsChunk(commandEnv), dropSaved, chunks) if manErr != nil { fmt.Fprintf(writer, "WARNING: re-manifestize failed: %v. Writing flat chunk list.\n", manErr) } else { @@ -203,12 +216,24 @@ func (c *commandFsDistributeChunks) Do(args []string, commandEnv *CommandEnv, wr entry.Chunks = finalChunks fmt.Fprintf(writer, "\nUpdating filer metadata (%d chunks moved)...\n", movedCount) + var stale bool err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { - return filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{ - Directory: dir, - Entry: entry, - }) + var uErr error + stale, uErr = updateEntryIfUnchanged(context.Background(), client, dir, entry, readEntry) + return uErr }) + if stale { + var copies []orphanedNeedle + for _, chunk := range slices.Concat(chunks, finalChunks) { + if fid := chunk.GetFileIdString(); !readFids[fid] { + readFids[fid] = true + copies = append(copies, orphanedNeedle{volumeId: chunk.GetFid().GetVolumeId(), fileId: fid}) + } + } + deleteOrphanedNeedles(commandEnv, util.FullPath(path), copies, true) + fmt.Fprintf(writer, "%s was changed or deleted during redistribution; the filer kept that version and the moved copies were deleted.\n", path) + return err + } if err != nil { fmt.Fprintf(writer, "FAILED to update filer metadata: %v\n", err) fmt.Fprintf(writer, "WARNING: New chunks were uploaded but metadata was not updated.\n") diff --git a/weed/shell/command_fs_merge_volumes.go b/weed/shell/command_fs_merge_volumes.go index 99b6140943..26059da18e 100644 --- a/weed/shell/command_fs_merge_volumes.go +++ b/weed/shell/command_fs_merge_volumes.go @@ -6,6 +6,7 @@ import ( "flag" "fmt" "io" + "math" "net/http" "sort" "strconv" @@ -21,6 +22,8 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util/wildcard" "github.com/seaweedfs/seaweedfs/weed/wdclient" "golang.org/x/exp/maps" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" "github.com/seaweedfs/seaweedfs/weed/operation" @@ -54,6 +57,9 @@ func (c *commandFsMergeVolumes) Help() string { -toVolumeId accepts a comma-separated list. With -fromVolumeId, the source chunks are distributed across the listed volumes by remaining capacity, so a volume that does not fit into any single target can still be cleared. + + -apply needs the shell lock (run "lock" first), so two runs never move the + same chunks at once. ` } @@ -72,6 +78,11 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer if err = fsMergeVolumesCommand.Parse(args); err != nil { return err } + if *apply { + if err = commandEnv.confirmIsLocked(args); err != nil { + return err + } + } dir := *dirArg if dir != "/" { @@ -171,7 +182,16 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer // look like mergeVolumes hadn't done anything. Track the old fids // and delete them below after the filer update commits, so the // filer never points at a fid we already deleted. - var movedSources []orphanedNeedle + var moved rewrittenNeedles + // The update only lands while the filer still holds the entry as + // the walk read it, so keep that copy before the first chunk is + // rewritten in place. + var readEntry *filer_pb.Entry + keepReadEntry := func() { + if readEntry == nil { + readEntry = proto.Clone(entry).(*filer_pb.Entry) + } + } for i, chunk := range entry.Chunks { if chunk.IsChunkManifest { if !c.manifestMayReferencePlan(plan, planCollections, needle.VolumeId(chunk.Fid.VolumeId)) { @@ -190,14 +210,18 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer if !changed || !*apply { continue } + keepReadEntry() entry.Chunks[i] = newChunk entryChanged = true - movedSources = append(movedSources, rewritten.sources...) + moved.sources = append(moved.sources, rewritten.sources...) // The old manifest needle is always orphaned when we // replace it with a freshly uploaded one, even when the // rewrite was triggered by sub-chunk moves rather than the // manifest volume itself being in the plan. - movedSources = append(movedSources, orphanedNeedle{volumeId: oldManifestVid, fileId: oldManifestFid}) + moved.sources = append(moved.sources, orphanedNeedle{volumeId: oldManifestVid, fileId: oldManifestFid}) + moved.targets = append(moved.targets, rewritten.targets...) + moved.targets = append(moved.targets, orphanedNeedle{volumeId: newChunk.Fid.VolumeId, fileId: newChunk.GetFileIdString()}) + moved.reservations = append(moved.reservations, rewritten.reservations...) continue } @@ -206,6 +230,7 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer continue } seen.record(chunkVolumeId) + keepReadEntry() oldFid := chunk.GetFileIdString() oldVid := chunk.Fid.VolumeId @@ -224,20 +249,29 @@ func (c *commandFsMergeVolumes) Do(args []string, commandEnv *CommandEnv, writer continue } entryChanged = true - movedSources = append(movedSources, orphanedNeedle{volumeId: oldVid, fileId: oldFid}) + moved.sources = append(moved.sources, orphanedNeedle{volumeId: oldVid, fileId: oldFid}) + moved.targets = append(moved.targets, orphanedNeedle{volumeId: uint32(toVolumeId), fileId: chunk.GetFileIdString()}) + moved.reservations = append(moved.reservations, reservation{src: chunkVolumeId, dst: toVolumeId, size: chunk.Size}) } if entryChanged { - if uErr := filer_pb.UpdateEntry(context.Background(), filerClient, &filer_pb.UpdateEntryRequest{ - Directory: string(parentPath), - Entry: entry, - }); uErr != nil { + stale, uErr := updateEntryIfUnchanged(context.Background(), filerClient, string(parentPath), entry, readEntry) + if stale { + fmt.Printf("skip %s: changed during merge\n", entryPath) + moved.targets = unreferencedTargets(context.Background(), filerClient, lookupFn, parentPath, entry.Name, moved.targets) + moved.abandon(commandEnv, entryPath, plan) + if len(entry.HardLinkId) > 0 { + processedHardLinks.Delete(string(entry.HardLinkId)) + } + return nil + } + if uErr != nil { fmt.Printf("failed to update %s: %v\n", entryPath, uErr) // Filer still references the source fids. Deleting them // now would lose data — abandon the cleanup for this // entry and let fsck reconcile later. return nil } - deleteOrphanedNeedles(commandEnv, entryPath, movedSources, false) + deleteOrphanedNeedles(commandEnv, entryPath, moved.sources, false) } return nil }) @@ -313,6 +347,61 @@ type rewrittenNeedles struct { reservations []reservation } +// abandon drops a rewrite the filer never took: the copies on the target +// volumes belong to nobody, and the plan gets back the room they held, or a +// multi-target merge keeps counting bytes it just deleted and skips later +// chunks for no reason. +func (r rewrittenNeedles) abandon(commandEnv *CommandEnv, entryPath util.FullPath, plan *mergePlan) { + deleteOrphanedNeedles(commandEnv, entryPath, r.targets, true) + for _, res := range r.reservations { + plan.release(res.src, res.dst, res.size) + } +} + +// updateEntryIfUnchanged writes entry back only while the filer still holds +// readEntry. An unconditional write would revert a change made since the read +// and the filer would delete the newer chunks as garbage. stale reports that +// the filer refused because the entry was changed or deleted, so nothing was +// written and the copies made for this update are referenced by nothing. +func updateEntryIfUnchanged(ctx context.Context, client filer_pb.SeaweedFilerClient, dir string, entry, readEntry *filer_pb.Entry) (stale bool, err error) { + err = filer_pb.UpdateEntry(ctx, client, &filer_pb.UpdateEntryRequest{ + Directory: dir, + Entry: entry, + Condition: &filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{ + {Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, ExpectedEntry: readEntry}, + }}, + }) + switch status.Code(err) { + case codes.FailedPrecondition, codes.NotFound: + return true, err + } + return false, err +} + +// unreferencedTargets drops the copies the filer's current entry points at. +// moveChunk keeps the key and cookie, so an overlapping run that moved the same +// chunk to the same volume wrote the same fid, and its committed update is what +// refused this one. When the entry cannot be read, no copy is reported. +func unreferencedTargets(ctx context.Context, client filer_pb.SeaweedFilerClient, lookupFn wdclient.LookupFileIdFunctionType, dir util.FullPath, name string, targets []orphanedNeedle) []orphanedNeedle { + resp, err := filer_pb.LookupEntry(ctx, client, &filer_pb.LookupDirectoryEntryRequest{Directory: string(dir), Name: name}) + if errors.Is(err, filer_pb.ErrNotFound) { + return targets + } + if err != nil { + return nil + } + chunks := resp.Entry.GetChunks() + dataChunks, manifestChunks, err := filer.ResolveChunkManifest(ctx, lookupFn, chunks, 0, math.MaxInt64, nil) + if err != nil { + return nil + } + referenced := make(map[string]bool) + for _, chunk := range slices.Concat(chunks, dataChunks, manifestChunks) { + referenced[chunk.GetFileIdString()] = true + } + return slices.DeleteFunc(targets, func(t orphanedNeedle) bool { return referenced[t.fileId] }) +} + // deleteOrphanedNeedles fans out BatchDelete RPCs to every replica of each // needle's volume. Errors are logged but never returned — the data is already // orphan at this point, so a failed cleanup just leaves work for a later fsck. @@ -740,16 +829,8 @@ func (c *commandFsMergeVolumes) rewriteManifestChunk( var rewritten rewrittenNeedles // Every path out of here that is not the successful return abandons the - // rewrite with the filer still pointing at the old manifest, so whatever - // already landed on the target volumes belongs to nobody. Give the plan - // back the room those copies were holding, or a multi-target merge keeps - // counting bytes it just deleted and skips later chunks for no reason. - abandon := func() { - deleteOrphanedNeedles(commandEnv, entryPath, rewritten.targets, true) - for _, r := range rewritten.reservations { - plan.release(r.src, r.dst, r.size) - } - } + // rewrite with the filer still pointing at the old manifest. + abandon := func() { rewritten.abandon(commandEnv, entryPath, plan) } anySubChanged := false for i, sub := range subChunks { if sub.IsChunkManifest { diff --git a/weed/shell/command_fs_merge_volumes_test.go b/weed/shell/command_fs_merge_volumes_test.go index d02dd60b71..37500e2d45 100644 --- a/weed/shell/command_fs_merge_volumes_test.go +++ b/weed/shell/command_fs_merge_volumes_test.go @@ -1,12 +1,22 @@ package shell import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "slices" "strings" "sync" "testing" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" ) func newMergeCmd(limitMB uint64, vols ...*master_pb.VolumeInformationMessage) *commandFsMergeVolumes { @@ -394,3 +404,112 @@ func TestWarnUnreferencedSources_ConcurrentRecording(t *testing.T) { t.Fatalf("expected no warning when all needles were seen, got %q", sb.String()) } } + +type conditionalUpdateFiler struct { + filer_pb.SeaweedFilerClient + err error + req *filer_pb.UpdateEntryRequest + current *filer_pb.Entry + lookupErr error + lookupReq *filer_pb.LookupDirectoryEntryRequest +} + +func (s *conditionalUpdateFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + s.lookupReq = req + if s.lookupErr != nil { + return nil, s.lookupErr + } + return &filer_pb.LookupDirectoryEntryResponse{Entry: s.current}, nil +} + +func (s *conditionalUpdateFiler) UpdateEntry(_ context.Context, req *filer_pb.UpdateEntryRequest, _ ...grpc.CallOption) (*filer_pb.UpdateEntryResponse, error) { + s.req = req + if s.err != nil { + return nil, s.err + } + return &filer_pb.UpdateEntryResponse{}, nil +} + +// The write-back is conditioned on the entry as it was read, and only a refusal +// that proves nothing was written counts as stale: the caller then deletes the +// copies it made, which would lose data if the update might have landed. +func TestUpdateEntryIfUnchanged(t *testing.T) { + read := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 3, FileKey: 1, Cookie: 7}}}} + moved := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{Fid: &filer_pb.FileId{VolumeId: 9, FileKey: 1, Cookie: 7}}}} + + for _, tc := range []struct { + name string + err error + stale bool + }{ + {"committed", nil, false}, + {"changed", status.Error(codes.FailedPrecondition, "precondition failed"), true}, + {"deleted", status.Error(codes.NotFound, "not found"), true}, + {"outcome unknown", status.Error(codes.Unavailable, "filer down"), false}, + } { + t.Run(tc.name, func(t *testing.T) { + fake := &conditionalUpdateFiler{err: tc.err} + stale, err := updateEntryIfUnchanged(context.Background(), fake, "/dir", moved, read) + if stale != tc.stale || (err != nil) != (tc.err != nil) { + t.Fatalf("stale=%v err=%v, want stale=%v err=%v", stale, err, tc.stale, tc.err) + } + clauses := fake.req.GetCondition().GetClauses() + if len(clauses) != 1 || clauses[0].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL || + !proto.Equal(clauses[0].ExpectedEntry, read) || !proto.Equal(fake.req.Entry, moved) { + t.Fatalf("request %v must write the moved entry conditioned on the read one", fake.req) + } + }) + } +} + +// A refused update may have been refused by an overlapping run that moved the +// same chunk to the same volume: moveChunk keeps the key and cookie, so that +// run's committed entry points at the very fid this run would delete. +func TestUnreferencedTargets(t *testing.T) { + committed := orphanedNeedle{volumeId: 9, fileId: "9,0101000007"} + own := orphanedNeedle{volumeId: 9, fileId: "9,0202000007"} + manifestCopy := orphanedNeedle{volumeId: 9, fileId: "9,0303000007"} + targets := func() []orphanedNeedle { return []orphanedNeedle{committed, own, manifestCopy} } + direct := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: committed.fileId}}} + + blob, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: []*filer_pb.FileChunk{{FileId: committed.fileId, Size: 10}}}) + 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) { + if fileId != manifestCopy.fileId { + return nil, fmt.Errorf("volume of %s unavailable", fileId) + } + return []string{volumeServer.URL + "/" + fileId}, nil + } + viaManifest := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: manifestCopy.fileId, IsChunkManifest: true, Size: 10}}} + unreadableManifest := &filer_pb.Entry{Name: "f", Chunks: []*filer_pb.FileChunk{{FileId: "9,0404000007", IsChunkManifest: true, Size: 10}}} + + for _, tc := range []struct { + name string + current *filer_pb.Entry + lookupErr error + want []orphanedNeedle + }{ + {"another run committed one copy", direct, nil, []orphanedNeedle{own, manifestCopy}}, + {"another run committed copies through a manifest", viaManifest, nil, []orphanedNeedle{own}}, + {"manifest unreadable", unreadableManifest, nil, nil}, + {"entry deleted", direct, status.Error(codes.NotFound, "not found"), targets()}, + {"entry unreadable", direct, status.Error(codes.Unavailable, "filer down"), nil}, + } { + t.Run(tc.name, func(t *testing.T) { + fake := &conditionalUpdateFiler{current: tc.current, lookupErr: tc.lookupErr} + got := unreferencedTargets(context.Background(), fake, lookupFn, "/dir", "f", targets()) + if !slices.Equal(got, tc.want) { + t.Fatalf("got %v, want %v", got, tc.want) + } + if fake.lookupReq.GetDirectory() != "/dir" || fake.lookupReq.GetName() != "f" { + t.Fatalf("looked up %v, want /dir f", fake.lookupReq) + } + }) + } +} diff --git a/weed/shell/command_fs_meta_change_volume_id.go b/weed/shell/command_fs_meta_change_volume_id.go index ef5a9386a2..23bedd5f99 100644 --- a/weed/shell/command_fs_meta_change_volume_id.go +++ b/weed/shell/command_fs_meta_change_volume_id.go @@ -12,6 +12,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/protobuf/proto" ) func init() { @@ -78,6 +79,7 @@ func (c *commandFsMetaChangeVolumeId) Do(args []string, commandEnv *CommandEnv, return filer_pb.TraverseBfs(context.Background(), commandEnv, util.FullPath(*dir), func(parentPath util.FullPath, entry *filer_pb.Entry) error { if !entry.IsDirectory { var hasChanges bool + var readEntry *filer_pb.Entry for _, chunk := range entry.Chunks { if chunk.IsChunkManifest { fmt.Printf("Change volume id for large file is not implemented yet: %s/%s\n", parentPath, entry.Name) @@ -85,6 +87,9 @@ func (c *commandFsMetaChangeVolumeId) Do(args []string, commandEnv *CommandEnv, } chunkVolumeId := chunk.Fid.VolumeId if toVolumeId, found := mapping[needle.VolumeId(chunkVolumeId)]; found { + if readEntry == nil { + readEntry = proto.Clone(entry).(*filer_pb.Entry) + } hasChanges = true chunk.Fid.VolumeId = uint32(toVolumeId) chunk.FileId = "" @@ -93,10 +98,9 @@ func (c *commandFsMetaChangeVolumeId) Do(args []string, commandEnv *CommandEnv, if hasChanges { println("Updating", parentPath, entry.Name) if *applyChanges { - if updateErr := filer_pb.UpdateEntry(context.Background(), client, &filer_pb.UpdateEntryRequest{ - Directory: string(parentPath), - Entry: entry, - }); updateErr != nil { + if stale, updateErr := updateEntryIfUnchanged(context.Background(), client, string(parentPath), entry, readEntry); stale { + fmt.Printf("skip %s/%s: changed since it was read\n", parentPath, entry.Name) + } else if updateErr != nil { fmt.Printf("failed to update %s/%s: %v\n", parentPath, entry.Name, updateErr) } } diff --git a/weed/shell/command_s3_versions_audit.go b/weed/shell/command_s3_versions_audit.go index 1be0b16ae1..c7b5aad164 100644 --- a/weed/shell/command_s3_versions_audit.go +++ b/weed/shell/command_s3_versions_audit.go @@ -13,6 +13,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" s3_constants "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/protobuf/proto" ) func init() { @@ -228,6 +229,7 @@ func healStrandedPointer(ctx context.Context, client filer_pb.SeaweedFilerClient if entry.Extended == nil { return nil } + readEntry := proto.Clone(entry).(*filer_pb.Entry) delete(entry.Extended, s3_constants.ExtLatestVersionIdKey) delete(entry.Extended, s3_constants.ExtLatestVersionFileNameKey) // Also clear the cached list metadata so a stale size/mtime/etag @@ -237,10 +239,9 @@ func healStrandedPointer(ctx context.Context, client filer_pb.SeaweedFilerClient delete(entry.Extended, s3_constants.ExtLatestVersionETagKey) delete(entry.Extended, s3_constants.ExtLatestVersionOwnerKey) delete(entry.Extended, s3_constants.ExtLatestVersionIsDeleteMarker) - _, err := client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{ - Directory: string(parentPath), - Entry: entry, - }) + // A version written since the audit read the directory has moved the + // pointer, and clearing it would hide that version. + _, err := updateEntryIfUnchanged(ctx, client, string(parentPath), entry, readEntry) return err } diff --git a/weed/shell/command_s3_versions_audit_test.go b/weed/shell/command_s3_versions_audit_test.go new file mode 100644 index 0000000000..b32c1051e4 --- /dev/null +++ b/weed/shell/command_s3_versions_audit_test.go @@ -0,0 +1,37 @@ +package shell + +import ( + "context" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + s3_constants "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" +) + +// The heal clears the pointer only while the directory still names the file +// the audit found missing; a version written since then has moved it. +func TestHealStrandedPointerIsConditional(t *testing.T) { + fake := &conditionalUpdateFiler{} + entry := &filer_pb.Entry{Name: "obj.versions", IsDirectory: true, Extended: map[string][]byte{ + s3_constants.ExtLatestVersionIdKey: []byte("v1"), + s3_constants.ExtLatestVersionFileNameKey: []byte("v_v1"), + "keep": []byte("x"), + }} + if err := healStrandedPointer(context.Background(), fake, "/buckets/b", entry); err != nil { + t.Fatalf("heal: %v", err) + } + + clauses := fake.req.GetCondition().GetClauses() + if len(clauses) != 1 || clauses[0].Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL { + t.Fatalf("heal must be conditioned on the audited entry, got %v", fake.req.GetCondition()) + } + if string(clauses[0].ExpectedEntry.Extended[s3_constants.ExtLatestVersionIdKey]) != "v1" { + t.Fatalf("expected entry must hold the pointer the audit saw, got %v", clauses[0].ExpectedEntry.Extended) + } + written := fake.req.Entry.Extended + _, idKept := written[s3_constants.ExtLatestVersionIdKey] + _, nameKept := written[s3_constants.ExtLatestVersionFileNameKey] + if idKept || nameKept || string(written["keep"]) != "x" { + t.Fatalf("written entry must clear only the pointer, got %v", written) + } +}