diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 2b19e0f96..d1bbf1450 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -236,6 +236,12 @@ func processDeleteEvent( glog.V(2).Infof("delete: %+v", resp) dest := toRemoteStorageLocation(util.FullPath(mountedDir), util.NewFullPath(resp.Directory, message.OldEntry.Name), remoteStorageMountLocation) if !message.OldEntry.IsDirectory { + // a deleted entry whose rename copy was still pending read its + // content from the source its marker names, which nothing else + // releases + if err := releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry); err != nil { + return err + } // The object at the key may have been replaced since the delete: a // rename copies another entry's content onto it, a rewrite uploads // over it, and a copy can land before its entry is visible. Delete @@ -267,6 +273,41 @@ func processDeleteEvent( return deleteRemoteFile(client, dest) } +// releasePendingSource deletes the source a pending-rename marker on entry +// names, once nothing needs it. The filer never deletes a marked source: a +// concurrent rename can copy onto that key before its entry is visible. So +// the object goes only while it is still the one entry's stamp describes, and +// no remote-only entry holds its path. +func releasePendingSource(filerSource filer_pb.FilerClient, client remote_storage.RemoteStorageClient, mountedDir string, remoteStorageMountLocation *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry) error { + src := util.FullPath(entry.GetExtended()[filer.ExtRemoteRenameSourceKey]) + if src == "" || !strings.HasPrefix(string(src), strings.TrimSuffix(mountedDir, "/")+"/") { + return nil + } + srcDest := toRemoteStorageLocation(util.FullPath(mountedDir), src, remoteStorageMountLocation) + srcDir, srcName := src.DirAndName() + current, err := currentEntry(filerSource, srcDir, srcName) + if err != nil { + return err + } + if isRemoteOnly(current) { + glog.V(0).Infof("keep %s: a remote-only entry holds the pending rename's source", remote_storage.FormatLocation(srcDest)) + return nil + } + live, err := client.StatFile(srcDest) + if errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { + return nil + } + if err != nil { + return err + } + if stamp := entry.GetRemoteEntry(); !describes(stamp, live) || stamp.RemoteETag != "" && live.RemoteETag == "" { + glog.V(0).Infof("keep %s: the pending rename's source changed since", remote_storage.FormatLocation(srcDest)) + return nil + } + glog.V(0).Infof("delete %s, the source of a pending rename", remote_storage.FormatLocation(srcDest)) + return deleteRemoteFile(client, srcDest) +} + // deleteRemoteFile deletes the object and treats an already-absent object as // deleted. func deleteRemoteFile(client remote_storage.RemoteStorageClient, dest *remote_pb.RemoteStorageLocation) error { @@ -340,7 +381,16 @@ func processUpdateEvent( return err } if isRemoteOnly(current) { - return completeRemoteOnlyRename(filerClient, client, message.NewParentPath, current, sourceOf(current, oldDest), dest, storageClass) + source := sourceOf(current, oldDest) + if err := completeRemoteOnlyRename(filerClient, client, message.NewParentPath, current, source, dest, storageClass); err != nil { + return err + } + // a pending entry the filer copied on this rename leaves the + // source its marker named + if pending := sourceOf(message.OldEntry, nil); pending != nil && !proto.Equal(pending, source) { + return releasePendingSource(filerSource, client, mountedDir, remoteStorageMountLocation, message.OldEntry) + } + return nil } if current == nil && len(message.NewEntry.Extended[filer.ExtRemoteRenameSourceKey]) > 0 { // the rename's copy was pending and the entry has moved on: a diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index 56df94cb9..8a6f5de69 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "io" + "slices" "strings" "testing" "time" @@ -1508,3 +1509,76 @@ func TestRenameOfAPendingEntryFollowsItsSource(t *testing.T) { } }) } + +func TestPendingRenameSourceIsReleasedOnlyWhileUnchanged(t *testing.T) { + const mountedDir = "/buckets" + mountLoc := &remote_pb.RemoteStorageLocation{Name: "b2", Bucket: "bucket", Path: "/"} + // b.jpg's copy failed after it replaced a target: it read a.jpg's object + pendingB := func() *filer_pb.Entry { + return &filer_pb.Entry{ + Name: "b.jpg", + RemoteEntry: &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6}, + Extended: map[string][]byte{filer.ExtRemoteRenameSourceKey: []byte("/buckets/b/a.jpg")}, + } + } + deleted := func() *filer_pb.SubscribeMetadataResponse { + return &filer_pb.SubscribeMetadataResponse{Directory: "/buckets/b", EventNotification: &filer_pb.EventNotification{OldEntry: pendingB()}} + } + deletedPaths := func(r *recordingRemote) (paths []string) { + for _, d := range r.deletes { + paths = append(paths, d.Path) + } + return + } + + t.Run("deleting the pending entry releases its source", func(t *testing.T) { + remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a")}, statETag: `"a"`} + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted()); err != nil { + t.Fatal(err) + } + if got := deletedPaths(remote); !slices.Contains(got, "/b/a.jpg") { + t.Errorf("deletes = %v, want the source", got) + } + }) + + t.Run("a source replaced since is kept", func(t *testing.T) { + // another rename copied onto a.jpg before its entry became visible + remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-d")}, statETag: `"d"`} + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted()); err != nil { + t.Fatal(err) + } + if got := deletedPaths(remote); slices.Contains(got, "/b/a.jpg") { + t.Errorf("deletes = %v, want the replaced source kept", got) + } + }) + + t.Run("a later rename the filer copied releases the source", func(t *testing.T) { + // b.jpg was renamed to c.jpg and the filer copied a.jpg there at once + copied := &filer_pb.Entry{Name: "c.jpg", Attributes: &filer_pb.FuseAttributes{FileSize: 6}, RemoteEntry: &filer_pb.RemoteEntry{StorageName: "b2", RemoteETag: `"a"`, RemoteSize: 6}} + remote := &recordingRemote{objects: map[string][]byte{"/b/a.jpg": []byte("data-a"), "/b/b.jpg": []byte("old-b"), "/b/c.jpg": []byte("data-a")}, statETag: `"a"`} + event := &filer_pb.SubscribeMetadataResponse{Directory: "/buckets/b", EventNotification: &filer_pb.EventNotification{ + OldEntry: pendingB(), NewParentPath: "/buckets/b", NewEntry: copied, + }} + filerClient := &pathFilerClient{entries: map[string]*filer_pb.Entry{"/buckets/b/c.jpg": copied}} + if err := processUpdateEvent(filerClient, filerClient, "", remote, mountedDir, mountLoc, event); err != nil { + t.Fatal(err) + } + if got := deletedPaths(remote); !slices.Contains(got, "/b/a.jpg") { + t.Errorf("deletes = %v, want the pending source released", got) + } + }) +} + +// pathFilerClient serves entries by path; a path it lacks is not found. +type pathFilerClient struct { + stubFilerClient + entries map[string]*filer_pb.Entry +} + +func (c *pathFilerClient) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest, _ ...grpc.CallOption) (*filer_pb.LookupDirectoryEntryResponse, error) { + return &filer_pb.LookupDirectoryEntryResponse{Entry: c.entries[req.Directory+"/"+req.Name]}, nil +} + +func (c *pathFilerClient) WithFilerClient(_ bool, fn func(filer_pb.SeaweedFilerClient) error) error { + return fn(c) +} diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index 2e5788eff..c55ddd46b 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/weed/server/filer_grpc_server_rename.go @@ -164,8 +164,6 @@ 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 @@ -237,9 +235,6 @@ func (event metadataEvent) notify(f *filer.Filer, ctx context.Context, signature if event.deleteOldRemote { 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) } @@ -397,7 +392,6 @@ 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, @@ -417,10 +411,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 - } + // the copy came from the pending source; the new key holds it now. + // The source itself is left to remote.sync, which deletes it only + // while it is still the object the rename event recorded newEntry.Extended = maps.Clone(newEntry.Extended) delete(newEntry.Extended, filer.ExtRemoteRenameSourceKey) } @@ -474,10 +467,9 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.Seawee }) } *metadataEvents = append(*metadataEvents, metadataEvent{ - oldEntry: sourceEntry, - newEntry: newEntry, - deleteOldRemote: copiedRemote != nil || deferredCopy, - deleteRemoteSource: pendingSource, + oldEntry: sourceEntry, + newEntry: newEntry, + deleteOldRemote: copiedRemote != nil || deferredCopy, }) 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 36f88f3f2..cb16a7a68 100644 --- a/weed/server/filer_grpc_server_rename_remote_test.go +++ b/weed/server/filer_grpc_server_rename_remote_test.go @@ -377,8 +377,9 @@ 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") + // the source is left to remote.sync: a concurrent rename may have copied + // onto that key before its entry is visible + assert.NotContains(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])