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

@@ -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))

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{