From a76ed8dd7a73c6b6620db497a43d7719c29f877c Mon Sep 17 00:00:00 2001 From: Alex Russell Date: Mon, 12 Oct 2026 05:29:26 +0200 Subject: [PATCH] filer: carry a deferred copy's stamp in the rename event; delete orphaned uploads - The deferred copy stamps the entry with a plain UpdateEntry, which publishes no event, so the queued rename event still carried the old key's stamp to peer filers and subscribers. Give it the copy's RemoteEntry before it is sent. The two notify loops share the code now. - remote.sync kept any object whose ETag changed since the delete. With no entry at the path that is an upload which finished after the delete, and keeping it leaves it on the remote for good. Look up the path first and keep a changed object only when an entry holds it. --- weed/command/filer_remote_sync_dir.go | 21 ++++----- weed/command/filer_remote_sync_dir_test.go | 17 +++++++- weed/server/filer_grpc_server_rename.go | 43 +++++++++++++------ .../filer_grpc_server_rename_remote_test.go | 2 + 4 files changed, 58 insertions(+), 25 deletions(-) diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index 85c987d20..8c660f70a 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -236,11 +236,20 @@ 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 { + current, err := currentEntry(filerSource, resp.Directory, message.OldEntry.Name) + if err != nil { + return err + } + if isRemoteOnly(current) { + glog.V(0).Infof("keep %s: a remote-only entry has taken its path since", remote_storage.FormatLocation(dest)) + return nil + } // 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. Match the recorded etag before deleting so only the - // object the delete meant goes away. - if oldETag := message.OldEntry.GetRemoteEntry().GetRemoteETag(); oldETag != "" { + // object the delete meant goes away. A changed object with no entry + // at the path is an upload that finished after the delete, and goes. + if oldETag := message.OldEntry.GetRemoteEntry().GetRemoteETag(); oldETag != "" && current != nil { if live, err := client.StatFile(dest); err == nil { if live.RemoteETag != "" && live.RemoteETag != oldETag { glog.V(0).Infof("keep %s: the object changed since the delete", remote_storage.FormatLocation(dest)) @@ -250,14 +259,6 @@ func processDeleteEvent( return err } } - current, err := currentEntry(filerSource, resp.Directory, message.OldEntry.Name) - if err != nil { - return err - } - if isRemoteOnly(current) { - glog.V(0).Infof("keep %s: a remote-only entry has taken its path since", remote_storage.FormatLocation(dest)) - return nil - } } if message.OldEntry.IsDirectory { glog.V(0).Infof("rmdir %s", 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 8267b9a2e..ba1e09122 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -1044,7 +1044,8 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse) deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 10} remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("copied")}, statETag: `"copy"`} - if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted); err != nil { + filerClient := &stubFilerClient{entry: &filer_pb.Entry{Name: "state.db.tmp", Content: []byte("copied")}} + if err := processDeleteEvent(filerClient, remote, mountedDir, mountLoc, deleted); err != nil { t.Fatalf("err = %v", err) } if len(remote.deletes) != 0 { @@ -1052,6 +1053,20 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { } }) + t.Run("a changed object with no entry at the path still goes", func(t *testing.T) { + // an upload that finished after the delete leaves a newer object that + // no entry describes; keeping it would leave it on the remote forever + deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse) + deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 10} + remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("late")}, statETag: `"late"`} + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted); err != nil { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 1 { + t.Errorf("deletes = %+v, want one", remote.deletes) + } + }) + t.Run("an unchanged object still goes", func(t *testing.T) { deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse) deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 10} diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go index af5123d14..418605e77 100644 --- a/weed/server/filer_grpc_server_rename.go +++ b/weed/server/filer_grpc_server_rename.go @@ -82,13 +82,8 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes) } fs.finishRemoteCopies(afterCommitCtx, pendingRemoteCopies) - for i := range metadataEvents { - event := &metadataEvents[i] - for _, pending := range pendingRemoteCopies { - if pending.failed && event.oldEntry.FullPath == pending.oldPath { - event.deleteOldRemote = false - } - } + applyRemoteCopies(metadataEvents, pendingRemoteCopies) + for _, event := range metadataEvents { event.notify(fs.filer, afterCommitCtx, req.Signatures) } @@ -154,13 +149,8 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, fs.filer.DeleteChunksNotRecursive(pendingChunkDeletes) } fs.finishRemoteCopies(afterCommitCtx, pendingRemoteCopies) - for i := range metadataEvents { - event := &metadataEvents[i] - for _, pending := range pendingRemoteCopies { - if pending.failed && event.oldEntry.FullPath == pending.oldPath { - event.deleteOldRemote = false - } - } + applyRemoteCopies(metadataEvents, pendingRemoteCopies) + for _, event := range metadataEvents { event.notify(fs.filer, afterCommitCtx, req.Signatures) } @@ -184,6 +174,7 @@ type pendingRemoteCopy struct { oldPath util.FullPath newPath util.FullPath failed bool + copied *filer_pb.RemoteEntry } // finishRemoteCopies runs the deferred remote copies after the commit. A @@ -208,6 +199,30 @@ func (fs *FilerServer) finishRemoteCopies(ctx context.Context, pendings []*pendi if updateErr := fs.filer.UpdateEntry(ctx, entry, entry, false); updateErr != nil { glog.WarningfCtx(ctx, "stamp remote rename copy of %s: %v", pending.newPath, updateErr) pending.failed = true + continue + } + pending.copied = remoteEntry + } +} + +// applyRemoteCopies brings the queued rename events in line with the deferred +// copies. The stamp is written with a plain UpdateEntry, which publishes no +// event, so the rename event carries it to peer filers and subscribers. A +// failed copy keeps the old object, which the entry still describes. +func applyRemoteCopies(events []metadataEvent, pendings []*pendingRemoteCopy) { + for i := range events { + event := &events[i] + for _, pending := range pendings { + if event.oldEntry.FullPath != pending.oldPath { + continue + } + if pending.failed { + event.deleteOldRemote = false + } else if pending.copied != nil && event.newEntry != nil { + newEntry := event.newEntry.ShallowClone() + newEntry.Remote = pending.copied + event.newEntry = newEntry + } } } } diff --git a/weed/server/filer_grpc_server_rename_remote_test.go b/weed/server/filer_grpc_server_rename_remote_test.go index e25d33610..ebbdf4e6a 100644 --- a/weed/server/filer_grpc_server_rename_remote_test.go +++ b/weed/server/filer_grpc_server_rename_remote_test.go @@ -147,6 +147,8 @@ func TestRenameRemoteOnlyEntryOverTargetKeepsTheCopy(t *testing.T) { require.Len(t, events, 2) assert.True(t, filer.IsMetadataOnlyDelete(events[0].notification.OldEntry), "the target's delete event must leave the remote object alone") assert.False(t, filer.IsMetadataOnlyDelete(events[1].notification.OldEntry)) + // the copy lands after the commit; the rename event still carries its stamp + assert.Equal(t, `"copy"`, events[1].notification.NewEntry.RemoteEntry.GetRemoteETag()) dst, err := store.FindEntry(context.Background(), "/buckets/b/dst/a.jpg") require.NoError(t, err) assert.Equal(t, uint64(101), dst.Attr.Inode)