filer: keep an old key a remote-only entry has retaken by cleanup time

A directory rename does not lock its children's paths, so between the
commit and the deferred delete of a copied child's old key another
rename can put a remote-only entry there, whose content is then only in
that object. Check the store before deleting.

weed/server/filer_grpc_server_rename.go

@@ -157,7 +157,12 @@ type metadataEvent struct {
 
 func (event metadataEvent) notify(f *filer.Filer, ctx context.Context, signatures []int32) {
 	if event.deleteOldRemote {
-		if err := f.DeleteFromRemote(ctx, event.oldEntry); err != nil {
+		// a child path is not locked by its directory's rename, so another
+		// rename may have put a remote-only entry at the old key meanwhile;
+		// the store is asked directly, a lazy fetch would revive the old key
+		if current, err := f.Store.FindEntry(ctx, event.oldEntry.FullPath); err == nil && current.IsInRemoteOnly() {
+			glog.V(0).InfofCtx(ctx, "keep remote object of renamed %s: a remote-only entry has taken its path", event.oldEntry.FullPath)
+		} else if err := f.DeleteFromRemote(ctx, event.oldEntry); err != nil {
 			glog.WarningfCtx(ctx, "delete remote object of renamed %s: %v", event.oldEntry.FullPath, err)
 		}
 	}

weed/server/filer_grpc_server_rename_remote_test.go

@@ -289,3 +289,22 @@ func TestRenameDirectoryCopiesRemoteOnlyChildren(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, `"orig"`, moved.Remote.RemoteETag)
 }
+
+func TestDeferredOldKeyDeleteKeepsAKeyRetakenSince(t *testing.T) {
+	client := copyingRenameRemoteClient{&renameRemoteClient{}}
+	server, store := newRenameRemoteTestServer(t, client)
+	event := metadataEvent{
+		oldEntry:        remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 101),
+		newEntry:        remoteOnlyRenameEntry("/buckets/b/moved/a.jpg", 101),
+		deleteOldRemote: true,
+	}
+
+	// another rename put a remote-only entry at the old path after the commit
+	store.entries["/buckets/b/src/a.jpg"] = remoteOnlyRenameEntry("/buckets/b/src/a.jpg", 303)
+	event.notify(server.filer, context.Background(), nil)
+	assert.Empty(t, client.recorded())
+
+	delete(store.entries, "/buckets/b/src/a.jpg")
+	event.notify(server.filer, context.Background(), nil)
+	assert.Equal(t, []string{"delete origin/src/a.jpg"}, client.recorded())
+}