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