From 29f9d0b80524f5928c10aa2ea4af414b1a65d41a Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 11 Oct 2026 19:35:58 -0700 Subject: [PATCH] remote.sync: match the delete event's etag before deleting the object A rename can copy another entry's content onto a key between the filer's check and the remote delete, and the delete then removes the copy. The delete event carries the object's etag; delete only when the live object still matches it. 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 | 14 ++++++++++ weed/command/filer_remote_sync_dir_test.go | 30 +++++++++++++++++++++- 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/weed/command/filer_remote_sync_dir.go b/weed/command/filer_remote_sync_dir.go index cd54393ac..85c987d20 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. Match the recorded etag before deleting so only the + // object the delete meant goes away. + 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 + } + } current, err := currentEntry(filerSource, resp.Directory, message.OldEntry.Name) if err != nil { return err diff --git a/weed/command/filer_remote_sync_dir_test.go b/weed/command/filer_remote_sync_dir_test.go index de9d92aed..8267b9a2e 100644 --- a/weed/command/filer_remote_sync_dir_test.go +++ b/weed/command/filer_remote_sync_dir_test.go @@ -686,6 +686,8 @@ type recordingRemote struct { // objects present on the remote, by path: StatFile and ReadFile answer // from it, everything else is ErrRemoteObjectNotFound. objects map[string][]byte + // statETag is the ETag StatFile reports for a present object. + statETag string } func (r *recordingRemote) WriteFile(loc *remote_pb.RemoteStorageLocation, entry *filer_pb.Entry, reader io.Reader) (*filer_pb.RemoteEntry, error) { @@ -711,7 +713,7 @@ func (r *recordingRemote) StatFile(loc *remote_pb.RemoteStorageLocation) (*filer if !ok { return nil, remote_storage.ErrRemoteObjectNotFound } - return &filer_pb.RemoteEntry{StorageName: loc.Name, RemoteSize: int64(len(body))}, nil + return &filer_pb.RemoteEntry{StorageName: loc.Name, RemoteSize: int64(len(body)), RemoteETag: r.statETag}, nil } func (r *recordingRemote) ReadFile(loc *remote_pb.RemoteStorageLocation, offset int64, size int64) ([]byte, error) { @@ -1036,6 +1038,32 @@ func TestDeleteEventAbsentRemoteObjectIsSuccess(t *testing.T) { } }) + t.Run("a replaced object is kept", func(t *testing.T) { + // the delete recorded the old object's etag; a rename copied another + // entry's content onto the key since, and it must not go with it + 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 { + t.Fatalf("err = %v", err) + } + if len(remote.deletes) != 0 { + t.Errorf("deletes = %+v, want none", 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} + 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) + } + if len(remote.deletes) != 1 { + t.Errorf("deletes = %+v, want one", 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{