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

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

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}

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
+			}
 		}
 	}
 }

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)