From 0a59ec11d6ddb5588e3d8b42ecf21f92d5e30d19 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 11 Oct 2026 20:38:11 -0700 Subject: [PATCH] remote.sync: delete only the object the event's stamp describes An object at a deleted key is not always the one the delete meant: a rename copy can land before its filer entry is visible, and a rewrite can upload over it. When the event recorded no remote stamp, or the live stat cannot confirm the recorded etag, the delete now keeps the object instead of treating it as a late upload. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/command/filer_remote_sync_dir.go | 31 ++++++++++------------ weed/command/filer_remote_sync_dir_test.go | 28 ++++++++++++++++++- 2 files changed, 41 insertions(+), 18 deletions(-) diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index a81a87b3d..3b0e60478 100644 --- a/weed/command/filer_remote_sync_dir.go +++ b/weed/command/filer_remote_sync_dir.go @@ -236,6 +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 { + // 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 + // only an object the event's stamp still describes. + if live, statErr := client.StatFile(dest); statErr == nil { + if oldRemote := message.OldEntry.GetRemoteEntry(); oldRemote == nil || + !describes(oldRemote, live) || + oldRemote.RemoteETag != "" && live.RemoteETag == "" { + glog.V(0).Infof("keep %s: the object changed since the delete", remote_storage.FormatLocation(dest)) + return nil + } + } else if !errors.Is(statErr, remote_storage.ErrRemoteObjectNotFound) { + return statErr + } current, err := currentEntry(filerSource, resp.Directory, message.OldEntry.Name) if err != nil { return err @@ -244,23 +258,6 @@ func processDeleteEvent( 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. That holds with no entry at the - // path too: a rename copies before its new entry becomes visible. An - // upload that finished after the delete is kept the same way, which - // leaks that object rather than risking the only copy of a rename. - if oldETag := message.OldEntry.GetRemoteEntry().GetRemoteETag(); oldETag != "" { - 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)) - return nil - } - } else if !errors.Is(err, remote_storage.ErrRemoteObjectNotFound) { - return err - } - } } 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 ae9d7f7e5..374963c37 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -1069,7 +1069,7 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { 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} + deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 3} remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("old")}, statETag: `"orig"`} if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted); err != nil { t.Fatalf("err = %v", err) @@ -1079,6 +1079,32 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { } }) + t.Run("a copy that landed before its entry is kept", func(t *testing.T) { + // the deleted entry was never replicated, so its event records no + // stamp to match the fresh rename copy waiting for its filer entry + remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("copied")}, statETag: `"copy"`} + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, resp); err != nil { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 0 { + t.Errorf("deletes = %+v, want none", remote.deletes) + } + }) + + t.Run("an unverifiable object is kept", func(t *testing.T) { + // the delete recorded an etag the live stat does not report; the + // object cannot be confirmed as the one the delete meant + deleted := proto.Clone(resp).(*filer_pb.SubscribeMetadataResponse) + deleted.EventNotification.OldEntry.RemoteEntry = &filer_pb.RemoteEntry{RemoteETag: `"orig"`, RemoteSize: 3} + remote := &recordingRemote{objects: map[string][]byte{"/b/vol/state.db.tmp": []byte("old")}} + if err := processDeleteEvent(&stubFilerClient{}, remote, mountedDir, mountLoc, deleted); err != nil { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 0 { + t.Errorf("deletes = %+v, want none", remote.deletes) + } + }) + t.Run("an entry with local data at the path does not stop the delete", func(t *testing.T) { remote := &recordingRemote{} filerClient := &stubFilerClient{entry: &filer_pb.Entry{