From 81e220e979804f3b28bb92f52badb52c825e6968 Mon Sep 17 00:00:00 2001 From: Alex Russell Date: Mon, 12 Oct 2026 06:21:57 +0200 Subject: [PATCH] filer, remote.sync: follow a pending rename's source across renames When a remote-only entry is renamed over another remote-only file and the deferred copy fails, it reads the source key through its marker. Renamed again the same way, the marker keeps naming that source, but: - remote.sync completed each rename from the event's old key. For the second one that is the first target's old object, which it copied over the destination before clearing the marker. It now copies from the source the entry's marker names. - For the first rename it found no entry at the new path and deleted the old key, which the later entry still reads. When the event's new entry carried a marker and the path is gone, it now leaves the key to the entry that moved on. - When the filer copies a pending entry immediately on a later rename, it now also releases the source the copy read, after the commit and with the same checks as the old key. --- weed/command/filer_remote_sync_dir.go | 24 +++++-- weed/command/filer_remote_sync_dir_test.go | 66 ++++++++++++++++++- weed/server/filer_grpc_server_rename.go | 51 +++++++++----- .../filer_grpc_server_rename_remote_test.go | 2 + 4 files changed, 119 insertions(+), 24 deletions(-) diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 2c30cfcf5..2b19e0f96 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -295,6 +295,15 @@ func processUpdateEvent( } oldDest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation) dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(message.NewParentPath, message.NewEntry.Name), remoteStorageMountLocation) + // sourceOf is where a remote-only entry's content is: the source its + // pending-rename marker names, which may be an earlier rename's old key, + // or else the event's old key + sourceOf := func(entry *filer_pb.Entry, eventSource *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation { + if src := string(entry.Extended[filer.ExtRemoteRenameSourceKey]); strings.HasPrefix(src, strings.TrimSuffix(mountedDir, "/")+"/") { + return toRemoteStorageLocation(util.FullPath(mountedDir), util.FullPath(src), remoteStorageMountLocation) + } + return eventSource + } if proto.Equal(oldDest, dest) && !shouldSendToRemote(message.NewEntry) { glog.V(2).Infof("skipping updating: %+v", resp) return nil @@ -331,7 +340,14 @@ func processUpdateEvent( return err } if isRemoteOnly(current) { - return completeRemoteOnlyRename(filerClient, client, message.NewParentPath, current, oldDest, dest, storageClass) + return completeRemoteOnlyRename(filerClient, client, message.NewParentPath, current, sourceOf(current, oldDest), dest, storageClass) + } + if current == nil && len(message.NewEntry.Extended[filer.ExtRemoteRenameSourceKey]) > 0 { + // the rename's copy was pending and the entry has moved on: a + // later rename carries the source in its own marker and finishes + // or releases it, so this one leaves it + glog.V(0).Infof("keep %s: the pending rename moved on", remote_storage.FormatLocation(oldDest)) + return nil } glog.V(0).Infof("delete %s", remote_storage.FormatLocation(oldDest)) if err := client.DeleteFile(oldDest); err != nil { @@ -347,7 +363,7 @@ func processUpdateEvent( if errors.Is(writeErr, errSuperseded) { glog.Errorf("skipping %s: %v", remote_storage.FormatLocation(dest), writeErr) if !proto.Equal(oldDest, dest) { - return uploadCurrentEntry(filerClient, filerSource, client, message.NewParentPath, message.NewEntry.Name, oldDest, dest, storageClass) + return uploadCurrentEntry(filerClient, filerSource, client, message.NewParentPath, message.NewEntry.Name, oldDest, dest, storageClass, sourceOf) } return nil } @@ -469,7 +485,7 @@ func openRemoteObject(client remote_storage.RemoteStorageClient, loc *remote_pb. // up once. A remote-only entry is finished the way completeRemoteOnlyRename // finishes a rename: its stamp can already describe the destination (a sync // plus remote.uncache in the meantime), and only then is it complete. -func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir, name string, oldDest, dest *remote_pb.RemoteStorageLocation, storageClass string) error { +func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, dir, name string, oldDest, dest *remote_pb.RemoteStorageLocation, storageClass string, sourceOf func(*filer_pb.Entry, *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation) error { current, _, _, err := filer_pb.GetEntry(context.Background(), filerSource, util.NewFullPath(dir, name)) if errors.Is(err, filer_pb.ErrNotFound) { return nil @@ -481,7 +497,7 @@ func uploadCurrentEntry(filerClient filer_pb.FilerClient, filerSource filer_pb.F return nil } if isRemoteOnly(current) { - return completeRemoteOnlyRename(filerClient, client, dir, current, oldDest, dest, storageClass) + return completeRemoteOnlyRename(filerClient, client, dir, current, sourceOf(current, oldDest), dest, storageClass) } if shouldSendToRemote(current) { glog.V(0).Infof("leaving %s to the rewrite that superseded the rename", remote_storage.FormatLocation(dest)) diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 69a96e551..56df94cb9 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -1377,7 +1377,7 @@ func TestSupersededRenameUploadsCurrentEntry(t *testing.T) { current := entryWith("b.txt", &filer_pb.RemoteEntry{StorageName: "gcs", RemoteMtime: 1786096669, RemoteSize: 1024}) remote := &recordingRemote{} filerClient := &stubFilerClient{entry: current} - err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", wantDelete, wantWrite, "") + err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", wantDelete, wantWrite, "", eventSource) if err == nil || !strings.Contains(err.Error(), "neither") { t.Errorf("err = %v, want the lost-content failure", err) } @@ -1390,7 +1390,7 @@ func TestSupersededRenameUploadsCurrentEntry(t *testing.T) { current := entryWith("b.txt", &filer_pb.RemoteEntry{StorageName: "gcs", RemoteMtime: 1786096669, RemoteSize: 7}) remote := &recordingRemote{objects: map[string][]byte{"/b/dir/b.txt": []byte("payload")}} filerClient := &stubFilerClient{entry: current} - if err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", wantDelete, wantWrite, ""); err != nil { + if err := uploadCurrentEntry(filerClient, filerClient, remote, "/buckets/b/dir", "b.txt", wantDelete, wantWrite, "", eventSource); err != nil { t.Fatalf("err = %v, want nil: the destination object matches the entry's stamp", err) } if len(remote.writes) != 0 { @@ -1446,3 +1446,65 @@ func TestCompleteRemoteOnlyRenameClearsThePendingSourceFirst(t *testing.T) { } }) } + +// eventSource resolves a rename's source to the event's old key. +func eventSource(_ *filer_pb.Entry, oldDest *remote_pb.RemoteStorageLocation) *remote_pb.RemoteStorageLocation { + return oldDest +} + +func TestRenameOfAPendingEntryFollowsItsSource(t *testing.T) { + // A was renamed over B and B over C, both copies failing: C reads A's + // object through the marker, and B's key holds the old target's bytes + const mountedDir = "/buckets" + mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/"} + remoteOnly := &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6} + pendingC := func() *filer_pb.Entry { + return &filer_pb.Entry{ + Name: "c.jpg", + Attributes: &filer_pb.FuseAttributes{FileSize: 6}, + RemoteEntry: remoteOnly, + Extended: map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/a.jpg")}, + } + } + event := func(oldName, newName string) *filer_pb.SubscribeMetadataResponse { + newEntry := pendingC() + newEntry.Name = newName + return &filer_pb.SubscribeMetadataResponse{ + Directory: "/buckets/b", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: oldName, RemoteEntry: remoteOnly}, + NewParentPath: "/buckets/b", + NewEntry: newEntry, + }, + } + } + objects := func() map[string][]byte { + return map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("old-b"), "/b/c.jpg": []byte("old-c")} + } + + t.Run("the first rename leaves the source to the entry that moved on", func(t *testing.T) { + remote := &recordingRemote{objects: objects()} + filerClient := &stubFilerClient{} // nothing at b.jpg any more + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event("a.jpg", "b.jpg")); err != nil { + t.Fatal(err) + } + if len(remote.deletes) != 0 || len(remote.writes) != 0 { + t.Errorf("deletes = %+v, writes = %+v, want neither: c.jpg still reads a.jpg", remote.deletes, remote.writes) + } + }) + + t.Run("the second rename copies from the marker's source", func(t *testing.T) { + current := pendingC() + remote := &recordingRemote{objects: objects()} + filerClient := &stubFilerClient{entry: current} + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event("b.jpg", "c.jpg")); err != nil { + t.Fatal(err) + } + if len(remote.written) != 1 || string(remote.written[0]) != "data-a" { + t.Errorf("written = %q, want a.jpg's content", remote.written) + } + if len(remote.deletes) != 1 || remote.deletes[0].Path != "/b/a.jpg" { + t.Errorf("deletes = %+v, want the source once the stamp cleared the marker", remote.deletes) + } + }) +} diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index 93d02a096..2e5788eff 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/weed/server/filer_grpc_server_rename.go @@ -164,6 +164,8 @@ type metadataEvent struct { // the old entry's remote object is deleted once the rename has committed: // it was copied to the new key, and a rename that fails must not lose it deleteOldRemote bool + // the pending source an immediate copy read, released along with it + deleteRemoteSource util.FullPath } // pendingRemoteCopy is a remote-only rename over a remote-only target whose @@ -233,25 +235,33 @@ func applyRemoteCopies(events []metadataEvent, pendings []*pendingRemoteCopy) { func (event metadataEvent) notify(f *filer.Filer, ctx context.Context, signatures []int32) { if event.deleteOldRemote { - // a child path is not locked by its directory's rename, so another - // rename may have put a remote-only entry at the old key meanwhile; - // the store is asked directly, a lazy fetch would revive the old key. - // When in doubt the object stays: an extra object loses nothing. - current, findErr := f.Store.FindEntry(ctx, event.oldEntry.FullPath) - switch { - case findErr != nil && !errors.Is(findErr, filer_pb.ErrNotFound): - glog.WarningfCtx(ctx, "keep remote object of renamed %s: look up its old path: %v", event.oldEntry.FullPath, findErr) - case findErr == nil && current.IsInRemoteOnly(): - glog.V(0).InfofCtx(ctx, "keep remote object of renamed %s: a remote-only entry has taken its path", event.oldEntry.FullPath) - default: - if err := f.DeleteFromRemote(ctx, event.oldEntry); err != nil { - glog.WarningfCtx(ctx, "delete remote object of renamed %s: %v", event.oldEntry.FullPath, err) - } - } + deleteRenamedRemoteObject(f, ctx, event.oldEntry) + } + if event.deleteRemoteSource != "" { + deleteRenamedRemoteObject(f, ctx, &filer.Entry{FullPath: event.deleteRemoteSource, Remote: event.oldEntry.Remote}) } f.NotifyUpdateEvent(ctx, event.oldEntry, event.newEntry, event.deleteChunks, false, signatures) } +// deleteRenamedRemoteObject deletes the remote object a committed rename no +// longer needs. A child path is not locked by its directory's rename, so +// another rename may have put a remote-only entry at the key meanwhile; the +// store is asked directly, a lazy fetch would revive the key. When in doubt +// the object stays: an extra object loses nothing. +func deleteRenamedRemoteObject(f *filer.Filer, ctx context.Context, entry *filer.Entry) { + current, findErr := f.Store.FindEntry(ctx, entry.FullPath) + switch { + case findErr != nil && !errors.Is(findErr, filer_pb.ErrNotFound): + glog.WarningfCtx(ctx, "keep remote object of renamed %s: look up its old path: %v", entry.FullPath, findErr) + case findErr == nil && current.IsInRemoteOnly(): + glog.V(0).InfofCtx(ctx, "keep remote object of renamed %s: a remote-only entry has taken its path", entry.FullPath) + default: + if err := f.DeleteFromRemote(ctx, entry); err != nil { + glog.WarningfCtx(ctx, "delete remote object of renamed %s: %v", entry.FullPath, err) + } + } +} + func (fs *FilerServer) moveEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32, skipTargetLookup bool, metadataEvents *[]metadataEvent, pendingChunkDeletes *[]*filer_pb.FileChunk, pendingRemoteCopies *[]*pendingRemoteCopy) error { if err := fs.moveSelfEntry(ctx, stream, oldParent, entry, newParent, newName, func() error { @@ -387,6 +397,7 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee } // add to new directory + var pendingSource util.FullPath newEntry := &filer.Entry{ FullPath: newPath, Attr: entry.Attr, @@ -407,6 +418,9 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee newEntry.Extended[filer.ExtRemoteRenameSourceKey] = []byte(fs.filer.RemoteSourcePath(entry, oldPath)) } else if copiedRemote != nil && len(newEntry.Extended[filer.ExtRemoteRenameSourceKey]) > 0 { // the copy came from the pending source; the new key holds it now + if source := fs.filer.RemoteSourcePath(entry, oldPath); source != oldPath { + pendingSource = source + } newEntry.Extended = maps.Clone(newEntry.Extended) delete(newEntry.Extended, filer.ExtRemoteRenameSourceKey) } @@ -460,9 +474,10 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee }) } *metadataEvents = append(*metadataEvents, metadataEvent{ - oldEntry: sourceEntry, - newEntry: newEntry, - deleteOldRemote: copiedRemote != nil || deferredCopy, + oldEntry: sourceEntry, + newEntry: newEntry, + deleteOldRemote: copiedRemote != nil || deferredCopy, + deleteRemoteSource: pendingSource, }) selfEvent := len(*metadataEvents) - 1 keepOldRemote := copiedRemote != nil || deferredCopy diff --git a/weed/server/filer_grpc_server_rename_remote_test.go b/weed/server/filer_grpc_server_rename_remote_test.go index 28ec86f15..36f88f3f2 100644 --- a/weed/server/filer_grpc_server_rename_remote_test.go +++ b/weed/server/filer_grpc_server_rename_remote_test.go @@ -377,6 +377,8 @@ func TestRenamePendingEntryAgainCopiesFromItsSource(t *testing.T) { client.copyErr = nil require.NoError(t, renameFile(server, "/buckets/b/dst/d.jpg", "/buckets/b/dst/c.jpg")) assert.Contains(t, client.recorded(), "copy origin/src/a.jpg origin/dst/c.jpg") + // the source the copy read goes too, once the rename has committed + assert.Contains(t, client.recorded(), "delete origin/src/a.jpg") c, err := store.FindEntry(context.Background(), "/buckets/b/dst/c.jpg") require.NoError(t, err) assert.Empty(t, c.Extended[filer.ExtRemoteRenameSourceKey])