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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* shell: check the heal clears the pointer file name too

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>

* shell: check unreferencedTargets reads the refused entry itself

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* shell: check unreferencedTargets keeps a copy a manifest references

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* shell: check unreferencedTargets keeps the manifest copy and skips an unreadable manifest

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

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")

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 {

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)
+			}
+		})
+	}
+}

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)
 						}
 					}

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
 }
 

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)
+	}
+}