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{