Fixed segfault -- we tried to close pq connection of failed tasks.

Task type was zerod, which means 'MOVE_PRIMARY', and we tried to cast it to
MovePartState and close its prev_conn, that is, close piece of invalid memory.

Author: arssher

src/copypart.c

@@ -381,18 +381,22 @@ init_cp_state(CopyPartState *cps)
  */
 static void finalize_cp_state(CopyPartState *cps)
 {
-	if (cps->src_conn != NULL)
-		reset_pqconn(&cps->src_conn);
-	if (cps->dst_conn != NULL)
-		reset_pqconn(&cps->dst_conn);
-	if (cps->type == COPYPARTTASK_MOVE_PRIMARY ||
-		cps->type == COPYPARTTASK_MOVE_REPLICA)
+	/* Failed tasks never open pq connections */
+	if (cps->res != TASK_FAILED)
 	{
-		MovePartState *mps = (MovePartState *) cps;
-		if (mps->prev_conn != NULL)
-			reset_pqconn(&mps->prev_conn);
-		if (mps->next_conn != NULL)
-			reset_pqconn(&mps->next_conn);
+		if (cps->src_conn != NULL)
+			reset_pqconn(&cps->src_conn);
+		if (cps->dst_conn != NULL)
+			reset_pqconn(&cps->dst_conn);
+		if (cps->type == COPYPARTTASK_MOVE_PRIMARY ||
+			cps->type == COPYPARTTASK_MOVE_REPLICA)
+		{
+			MovePartState *mps = (MovePartState *) cps;
+			if (mps->prev_conn != NULL)
+				reset_pqconn(&mps->prev_conn);
+			if (mps->next_conn != NULL)
+				reset_pqconn(&mps->next_conn);
+		}
 	}
 }
 

src/shard.c

@@ -287,9 +287,12 @@ set_replevel(Cmd *cmd)
 			if (rc.count < replevel)
 			{
 				CreateReplicaState *crs = palloc0(sizeof(CreateReplicaState));
-				init_cr_state(crs, rc.part_name, workers[rand() % num_workers]);
+				int32 dst_node = workers[rand() % num_workers];
+				init_cr_state(crs, rc.part_name, dst_node);
 				tasks[ntasks] = (CopyPartState *) crs;
 				ntasks++;
+				shmn_elog(DEBUG1, "Adding replica for shard %s on node %d",
+						  rc.part_name, dst_node);
 			}
 		}