shd/symlink: soft links entry recreate fails

While an entry heal is on going, if there is a pending
metadata set on the source dentry, then the recreation
of soft link will fail to create on the destination.

This patch introduce a new flag and avoid overloading
the newentry mark flag

Change-Id: I862ab8d4318746ef2b15823ef5ad2272bff8aed7
Fixes: gluster#4065
Signed-off-by: Mohammed Rafi KC <[email protected]>

tests/bugs/replicate/issue-4064-softlink-entry-recreation.t

@@ -0,0 +1,45 @@
+#!/bin/bash
+
+. $(dirname $0)/../../include.rc
+. $(dirname $0)/../../volume.rc
+. $(dirname $0)/../../afr.rc
+cleanup;
+
+TEST glusterd
+TEST pidof glusterd
+TEST $CLI volume create $V0 replica 3 $H0:$B0/${V0}{0,1,2};
+TEST $CLI volume set $V0 cluster.self-heal-daemon off
+TEST $CLI volume start $V0
+
+TEST glusterfs --volfile-id=/$V0 --volfile-server=$H0 $M0 --attribute-timeout=0 --entry-timeout=0
+EXPECT_WITHIN $PROCESS_UP_TIMEOUT "1" afr_child_up_status $V0 0
+EXPECT_WITHIN $PROCESS_UP_TIMEOUT "1" afr_child_up_status $V0 1
+EXPECT_WITHIN $PROCESS_UP_TIMEOUT "1" afr_child_up_status $V0 2
+echo "Data">$M0/FILE
+ret=$?
+TEST [ $ret -eq 0 ]
+
+TEST kill_brick $V0 $H0 $B0/${V0}2
+
+TEST ln -s $M0/FILE $M0/SOFT
+TEST ln  $M0/FILE $M0/HARD
+#hardlink to a softlink
+TEST ln $M0/SOFT $M0/SOFTHARD
+
+#Set a metadata heal on the softlink
+TEST chown -h root:root $M0/SOFT
+
+#start the brick
+TEST $CLI volume start $V0 force
+
+EXPECT_WITHIN $PROCESS_UP_TIMEOUT "1" brick_up_status $V0 $H0 $B0/${V0}2
+
+TEST $CLI volume set $V0 cluster.self-heal-daemon on
+EXPECT_WITHIN $CHILD_UP_TIMEOUT "1" afr_child_up_status_in_shd $V0 0
+EXPECT_WITHIN $CHILD_UP_TIMEOUT "1" afr_child_up_status_in_shd $V0 1
+EXPECT_WITHIN $CHILD_UP_TIMEOUT "1" afr_child_up_status_in_shd $V0 2
+TEST $CLI volume heal $V0
+EXPECT_WITHIN $HEAL_TIMEOUT "^0$" get_pending_heal_count $V0
+
+EXPECT_WITHIN $UMOUNT_TIMEOUT "Y" force_umount $M0
+cleanup;

xlators/cluster/afr/src/afr-self-heal-entry.c

@@ -181,10 +181,12 @@ afr_selfheal_entry_delete(xlator_t *this, inode_t *dir, const char *name,
 static int
 afr_new_entry_mark_status(call_frame_t *frame, loc_t *loc,
                           struct afr_reply *lookup_replies,
-                          unsigned char *sources, int source, int dst)
+                          unsigned char *sources, int source, int dst,
+                          gf_boolean_t *dst_hardlink)
 {
     xlator_t *this = frame->this;
     afr_private_t *priv = this->private;
+    struct iatt *iatt = NULL;
     int pending = 0;
     int metadata_idx = 0;
     int idx = -1;
@@ -194,6 +196,7 @@ afr_new_entry_mark_status(call_frame_t *frame, loc_t *loc,
     if (source == -1) {
         goto lookup;
     }
+    iatt = &lookup_replies[source].poststat;
 
     if (IA_ISDIR(lookup_replies[source].poststat.ia_type)) {
         goto lookup;
@@ -230,12 +233,25 @@ afr_new_entry_mark_status(call_frame_t *frame, loc_t *loc,
             goto lookup;
         }
     }
+    if (iatt->ia_type == IA_IFLNK && dst_hardlink) {
+        ret = syncop_lookup(priv->children[dst], loc, 0, 0, 0, 0);
+        if (ret == 0) {
+            *dst_hardlink = _gf_true;
+        }
+        /* If it is a soft link, we need to check if a hardlink to
+         * this softlink present in the dst, hence we perform a
+         * lookup here*/
+    }
     /*Pending is marked on all source bricks, we definitely know that new entry
      * marking is not needed*/
     return 0;
 
 lookup:
     ret = syncop_lookup(priv->children[dst], loc, 0, 0, 0, 0);
+    if (ret == 0 && dst_hardlink) {
+        *dst_hardlink = _gf_true;
+    }
+
     return ret;
 }
 
@@ -267,6 +283,7 @@ afr_selfheal_recreate_entry(call_frame_t *frame, int dst, int source,
     unsigned char *newentry = NULL;
     char iatt_uuid_str[64] = {0};
     char dir_uuid_str[64] = {0};
+    gf_boolean_t dst_hardlink = _gf_false;
 
     priv = this->private;
     iatt = &replies[source].poststat;
@@ -302,7 +319,7 @@ afr_selfheal_recreate_entry(call_frame_t *frame, int dst, int source,
     srcloc.inode = inode_ref(inode);
     gf_uuid_copy(srcloc.gfid, iatt->ia_gfid);
     ret = afr_new_entry_mark_status(frame, &srcloc, replies, sources, source,
-                                    dst);
+                                    dst, &dst_hardlink);
     if (ret == -ENOENT || ret == -ESTALE) {
         newentry[dst] = 1;
         ret = afr_selfheal_newentry_mark(frame, this, inode, source, replies,
@@ -329,7 +346,7 @@ afr_selfheal_recreate_entry(call_frame_t *frame, int dst, int source,
             ret = syncop_mkdir(priv->children[dst], &loc, mode, 0, xdata, NULL);
             break;
         case IA_IFLNK:
-            if (!newentry[dst]) {
+            if (dst_hardlink) {
                 ret = syncop_link(priv->children[dst], &srcloc, &loc, &newent,
                                   NULL, NULL);
             } else {