diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go index c922301c6..ef308121e 100644 --- a/weed/command/filer_sync.go +++ b/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 } diff --git a/weed/command/filer_sync_process_test.go b/weed/command/filer_sync_process_test.go index 11313a60c..3f0d35908 100644 --- a/weed/command/filer_sync_process_test.go +++ b/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) { diff --git a/weed/replication/replicator.go b/weed/replication/replicator.go index ceac94570..26915d5c7 100644 --- a/weed/replication/replicator.go +++ b/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) diff --git a/weed/replication/replicator_test.go b/weed/replication/replicator_test.go index eb608550f..6c4947ef5 100644 --- a/weed/replication/replicator_test.go +++ b/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) { diff --git a/weed/replication/sink/filersink/filer_sink.go b/weed/replication/sink/filersink/filer_sink.go index 340b3d4eb..f02761268 100644 --- a/weed/replication/sink/filersink/filer_sink.go +++ b/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 { diff --git a/weed/replication/sink/replication_sink.go b/weed/replication/sink/replication_sink.go index 29e6bbf8c..80eca0094 100644 --- a/weed/replication/sink/replication_sink.go +++ b/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 )