feat(filer.sync): move filer-sink renames natively via AtomicRenameEntry

create-then-delete is unsafe for the filer sink: CreateEntry returns nil
without creating on a transient chunk-copy error, so the paired delete could
remove the only valid destination copy; a directory rename also deleted the
old subtree before descendants were recreated, and left old chunks behind.

Add an optional EntryMover sink capability and implement it on the filer sink
via AtomicRenameEntry — one atomic, metadata-only move that relocates a whole
subtree in a single transaction. Renames prefer it; sinks without a native
move keep create-then-delete. When the old path is already gone (a descendant
the parent rename moved, or one never replicated) MoveEntry creates the new
path instead, re-checking existence with a lookup so a rolled-back move that
left the old entry intact is retried rather than mistaken for gone.

weed/command/filer_sync.go

@@ -612,22 +612,30 @@ func genProcessFunction(sourcePath string, targetPath string, excludePaths []str
 			if util.IsEqualOrUnder(string(sourceNewKey), sourcePath) {
 				// new key is also in the watched directory
 				if filer_pb.IsRename(resp) {
-					// A real move: UpdateEntry cannot move an entry, so create at the new
-					// key first, then delete the old (when deletes are enabled), so a crash
-					// between the two leaves the entry visible rather than lost.
 					newKey := buildKey(dataSink, message, targetPath, sourceNewKey, sourcePath)
-					if err := dataSink.CreateEntry(newKey, message.NewEntry, message.Signatures); err != nil {
-						return fmt.Errorf("create entry2 : %w", err)
-					}
-					// derive the old key the same way as the new one so the delete
-					// targets the same sink-side normalization the create used. Guard
-					// against the watched root itself moving, where the old key would
-					// resolve to targetPath and recursively delete the whole sink tree.
+					// With deletes enabled a rename relocates the entry. Guard the
+					// watched root itself, whose old key would resolve to the target
+					// root and recursively delete the whole sink tree.
 					if doDeleteFiles && string(sourceOldKey) != sourcePath {
 						oldKey := buildKey(dataSink, message, targetPath, sourceOldKey, sourcePath)
+						if mover, ok := dataSink.(sink.EntryMover); ok {
+							// native atomic move: no re-copy, no descendant gap, no chunk leak.
+							return mover.MoveEntry(oldKey, newKey, message.NewEntry, message.Signatures)
+						}
+						// no native move: create the new entry first, then delete the
+						// old, so a crash between the two leaves the entry visible.
+						if err := dataSink.CreateEntry(newKey, message.NewEntry, message.Signatures); err != nil {
+							return fmt.Errorf("create entry2 : %w", err)
+						}
 						if err := dataSink.DeleteEntry(oldKey, message.OldEntry.IsDirectory, false, message.Signatures); err != nil {
 							return fmt.Errorf("delete old entry %v: %w", oldKey, err)
 						}
+						return nil
+					}
+					// deletes disabled (backup/incremental) or the watched root moved:
+					// create the new entry and keep the old.
+					if err := dataSink.CreateEntry(newKey, message.NewEntry, message.Signatures); err != nil {
+						return fmt.Errorf("create entry2 : %w", err)
 					}
 					return nil
 				}

weed/command/filer_sync_process_test.go

@@ -58,6 +58,23 @@ func (s *recordingSyncSink) GetSinkToDirectory() string         { return "/dest"
 func (s *recordingSyncSink) SetSourceFiler(*source.FilerSource) {}
 func (s *recordingSyncSink) IsIncremental() bool                { return false }
 
+// movingSyncSink is a recordingSyncSink that also implements sink.EntryMover,
+// modeling a sink (like the filer) with a native atomic move.
+type movingSyncSink struct {
+	*recordingSyncSink
+	moveOldKeys []string
+	moveNewKeys []string
+}
+
+var _ sink.EntryMover = (*movingSyncSink)(nil)
+
+func (s *movingSyncSink) MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry, signatures []int32) error {
+	s.moveOldKeys = append(s.moveOldKeys, oldKey)
+	s.moveNewKeys = append(s.moveNewKeys, newKey)
+	s.ordered = append(s.ordered, "move")
+	return nil
+}
+
 func TestPathIsEqualOrUnderUsesDirectoryBoundaries(t *testing.T) {
 	tests := []struct {
 		name      string
@@ -125,6 +142,57 @@ func TestGenProcessFunctionRenameCreatesThenDeletes(t *testing.T) {
 	}
 }
 
+// A sink with a native move relocates a rename via MoveEntry, not create-then-delete.
+func TestGenProcessFunctionRenameUsesMoveEntryWhenSupported(t *testing.T) {
+	dataSink := &movingSyncSink{recordingSyncSink: &recordingSyncSink{}}
+	processFn := genProcessFunction("/foo", "/dest", nil, nil, nil, nil, dataSink, true, false)
+
+	err := processFn(&filer_pb.SubscribeMetadataResponse{
+		Directory: "/foo/dir",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      &filer_pb.Entry{Name: "old.txt"},
+			NewEntry:      &filer_pb.Entry{Name: "new.txt"},
+			NewParentPath: "/foo/dir",
+		},
+	})
+	if err != nil {
+		t.Fatalf("processFn rename via mover: %v", err)
+	}
+
+	if len(dataSink.moveOldKeys) != 1 || dataSink.moveOldKeys[0] != "/dest/dir/old.txt" || dataSink.moveNewKeys[0] != "/dest/dir/new.txt" {
+		t.Fatalf("move old=%v new=%v, want one move /dest/dir/old.txt => /dest/dir/new.txt", dataSink.moveOldKeys, dataSink.moveNewKeys)
+	}
+	if len(dataSink.createKeys) != 0 || len(dataSink.deleteKeys) != 0 || len(dataSink.updateKeys) != 0 {
+		t.Fatalf("native move must not create/delete/update: creates=%v deletes=%v updates=%v", dataSink.createKeys, dataSink.deleteKeys, dataSink.updateKeys)
+	}
+}
+
+// With deletes disabled (backup/incremental), even a mover keeps the old entry:
+// it creates the new one and does not move or delete the old.
+func TestGenProcessFunctionRenameMoverKeepsOldWhenDeletesDisabled(t *testing.T) {
+	dataSink := &movingSyncSink{recordingSyncSink: &recordingSyncSink{}}
+	processFn := genProcessFunction("/foo", "/dest", nil, nil, nil, nil, dataSink, false, false)
+
+	err := processFn(&filer_pb.SubscribeMetadataResponse{
+		Directory: "/foo/dir",
+		EventNotification: &filer_pb.EventNotification{
+			OldEntry:      &filer_pb.Entry{Name: "old.txt"},
+			NewEntry:      &filer_pb.Entry{Name: "new.txt"},
+			NewParentPath: "/foo/dir",
+		},
+	})
+	if err != nil {
+		t.Fatalf("processFn rename (no delete) via mover: %v", err)
+	}
+
+	if len(dataSink.createKeys) != 1 || dataSink.createKeys[0] != "/dest/dir/new.txt" {
+		t.Fatalf("create keys = %v, want [/dest/dir/new.txt]", dataSink.createKeys)
+	}
+	if len(dataSink.moveOldKeys) != 0 || len(dataSink.deleteKeys) != 0 {
+		t.Fatalf("deletes-disabled rename must keep old: moves=%v deletes=%v", dataSink.moveOldKeys, dataSink.deleteKeys)
+	}
+}
+
 // An in-place update (same dir + same name) must route to UpdateEntry, never the
 // rename create-then-delete path — otherwise it would delete the key it just wrote.
 func TestGenProcessFunctionInPlaceUpdateUsesUpdateEntry(t *testing.T) {

weed/replication/replicator.go

@@ -112,10 +112,14 @@ func (r *Replicator) Replicate(ctx context.Context, key string, message *filer_p
 	}
 
 	if oldSinkKey != newSinkKey {
-		// A real move: the path changed. UpdateEntry can only mutate an entry in place
-		// at its existing path; it cannot move it. Create at the new key first, then
-		// delete the old, so a crash between the two leaves the entry visible under
-		// both names rather than gone.
+		// A real move: the path changed. UpdateEntry cannot move an entry.
+		if mover, ok := r.sink.(sink.EntryMover); ok {
+			glog.V(4).Infof("moving %v => %v", oldSinkKey, newSinkKey)
+			return mover.MoveEntry(oldSinkKey, newSinkKey, newEntry, message.Signatures)
+		}
+		// Sinks without a native move: create at the new key first, then delete the
+		// old, so a crash between the two leaves the entry visible under both names
+		// rather than gone.
 		glog.V(4).Infof("creating renamed %v", newSinkKey)
 		if err := r.sink.CreateEntry(newSinkKey, newEntry, message.Signatures); err != nil {
 			return fmt.Errorf("create renamed entry %v: %w", newSinkKey, err)

weed/replication/replicator_test.go

@@ -77,6 +77,26 @@ func (s *recordingSink) IsIncremental() bool {
 	return s.incremental
 }
 
+type moveCall struct {
+	oldKey string
+	newKey string
+}
+
+// movingSink is a recordingSink that also implements sink.EntryMover, modeling
+// a sink (like the filer) with a native atomic move.
+type movingSink struct {
+	*recordingSink
+	moveCalls []moveCall
+}
+
+var _ sink.EntryMover = (*movingSink)(nil)
+
+func (s *movingSink) MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry, signatures []int32) error {
+	s.moveCalls = append(s.moveCalls, moveCall{oldKey: oldKey, newKey: newKey})
+	s.ordered = append(s.ordered, "move")
+	return nil
+}
+
 func TestReplicateRenameUsesTargetKeyForNonFilerSink(t *testing.T) {
 	s := &recordingSink{name: "local", sinkToDirectory: "/dest"}
 	r := &Replicator{
@@ -120,9 +140,10 @@ func TestReplicateRenameUsesTargetKeyForNonFilerSink(t *testing.T) {
 // A real move to a filer sink also goes through create-then-delete (the
 // filer-sink special-case that previously routed renames to UpdateEntry is
 // gone): UpdateEntry cannot move an entry across paths.
-func TestReplicateRenameUsesCreateThenDeleteForFilerSink(t *testing.T) {
+// A sink without a native move falls back to create-then-delete for a rename.
+func TestReplicateRenameWithoutMoverUsesCreateThenDelete(t *testing.T) {
 	s := &recordingSink{
-		name:                "filer",
+		name:                "s3",
 		sinkToDirectory:     "/dest",
 		updateFoundExisting: true,
 	}
@@ -151,7 +172,7 @@ func TestReplicateRenameUsesCreateThenDeleteForFilerSink(t *testing.T) {
 	}
 
 	if len(s.updateCalls) != 0 {
-		t.Fatalf("expected filer-sink rename to bypass UpdateEntry, got %d calls", len(s.updateCalls))
+		t.Fatalf("expected rename to bypass UpdateEntry, got %d calls", len(s.updateCalls))
 	}
 	if len(s.createCalls) != 1 || s.createCalls[0].key != "/dest/new/renamed.txt" {
 		t.Fatalf("create calls = %+v, want target sink key", s.createCalls)
@@ -164,6 +185,32 @@ func TestReplicateRenameUsesCreateThenDeleteForFilerSink(t *testing.T) {
 	}
 }
 
+// A sink with a native move (the filer) relocates a rename via MoveEntry — never
+// create-then-delete, so a failed copy can't delete the only valid destination.
+func TestReplicateRenameUsesMoveEntryWhenSupported(t *testing.T) {
+	s := &movingSink{recordingSink: &recordingSink{name: "filer", sinkToDirectory: "/dest"}}
+	r := &Replicator{
+		sink:   s,
+		source: &source.FilerSource{Dir: "/source"},
+	}
+
+	err := r.Replicate(context.Background(), "/source/old/file.txt", &filer_pb.EventNotification{
+		OldEntry:      &filer_pb.Entry{Name: "file.txt"},
+		NewEntry:      &filer_pb.Entry{Name: "renamed.txt"},
+		NewParentPath: "/source/new",
+	})
+	if err != nil {
+		t.Fatalf("Replicate rename: %v", err)
+	}
+
+	if len(s.moveCalls) != 1 || s.moveCalls[0].oldKey != "/dest/old/file.txt" || s.moveCalls[0].newKey != "/dest/new/renamed.txt" {
+		t.Fatalf("move calls = %+v, want one move /dest/old/file.txt => /dest/new/renamed.txt", s.moveCalls)
+	}
+	if len(s.createCalls) != 0 || len(s.deleteCalls) != 0 || len(s.updateCalls) != 0 {
+		t.Fatalf("native move must not create/delete/update: creates=%v deletes=%v updates=%v", s.createCalls, s.deleteCalls, s.updateCalls)
+	}
+}
+
 // An in-place update (same parent + same name, so oldSinkKey == newSinkKey)
 // still routes to UpdateEntry rather than create-then-delete.
 func TestReplicateInPlaceUpdateUsesUpdateEntry(t *testing.T) {

weed/replication/sink/filersink/filer_sink.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"math"
+	"strings"
 	"sync"
 
 	"github.com/seaweedfs/seaweedfs/weed/operation"
@@ -166,6 +167,67 @@ func (fs *FilerSink) DeleteEntry(key string, isDirectory, deleteIncludeChunks bo
 	return nil
 }
 
+var _ sink.EntryMover = (*FilerSink)(nil)
+
+// MoveEntry relocates oldKey to newKey on the target filer via AtomicRenameEntry:
+// a metadata-only move that relocates a whole subtree in one transaction, so a
+// directory rename never leaves descendants missing and chunks are neither
+// re-copied nor leaked.
+//
+// When the move fails because the old path is genuinely gone on the sink — a
+// descendant the parent rename already relocated, or one never replicated —
+// there is nothing to move, so it creates the new path instead (CreateEntry
+// short-circuits when the entry is already there, and never deletes). Existence
+// is re-checked with a direct lookup rather than inferred from the rename error,
+// so a rolled-back move that left the old entry intact propagates for retry
+// instead of being mistaken for "gone".
+func (fs *FilerSink) MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry, signatures []int32) error {
+	oldDir, oldName := util.FullPath(oldKey).DirAndName()
+	newDir, newName := util.FullPath(newKey).DirAndName()
+
+	err := fs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		_, err := client.AtomicRenameEntry(context.Background(), &filer_pb.AtomicRenameEntryRequest{
+			OldDirectory: oldDir,
+			OldName:      oldName,
+			NewDirectory: newDir,
+			NewName:      newName,
+			Signatures:   signatures,
+		})
+		return err
+	})
+	if err == nil {
+		return nil
+	}
+	if missing, lookupErr := fs.entryMissing(oldKey); lookupErr == nil && missing {
+		glog.V(2).Infof("move %s => %s: old path gone, creating %s", oldKey, newKey, newKey)
+		return fs.CreateEntry(newKey, newEntry, signatures)
+	}
+	return fmt.Errorf("move %s => %s: %w", oldKey, newKey, err)
+}
+
+// entryMissing reports whether key has no entry on the target filer. A lookup
+// not-found (sentinel or the gRPC string form) means missing; any other lookup
+// error is returned so the caller does not treat an unknown state as missing.
+func (fs *FilerSink) entryMissing(key string) (bool, error) {
+	dir, name := util.FullPath(key).DirAndName()
+	missing := false
+	err := fs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
+		_, lookupErr := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{
+			Directory: dir,
+			Name:      name,
+		})
+		if lookupErr == nil {
+			return nil
+		}
+		if errors.Is(lookupErr, filer_pb.ErrNotFound) || strings.Contains(lookupErr.Error(), filer_pb.ErrNotFound.Error()) {
+			missing = true
+			return nil
+		}
+		return lookupErr
+	})
+	return missing, err
+}
+
 func (fs *FilerSink) CreateEntry(key string, entry *filer_pb.Entry, signatures []int32) error {
 
 	return fs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {

weed/replication/sink/replication_sink.go

@@ -17,6 +17,15 @@ type ReplicationSink interface {
 	IsIncremental() bool
 }
 
+// EntryMover is an optional capability for sinks that can relocate an entry
+// natively, in one atomic step, instead of create-then-delete. Drivers prefer
+// it for a rename so a failed copy can never leave the source deleted with no
+// committed destination, a directory move never deletes descendants before they
+// are recreated, and the entry's chunks are neither re-copied nor leaked.
+type EntryMover interface {
+	MoveEntry(oldKey, newKey string, newEntry *filer_pb.Entry, signatures []int32) error
+}
+
 var (
 	Sinks []ReplicationSink
 )