diff --git a/crmd/crmd_lrm.h b/crmd/crmd_lrm.h

index 46cddc2..3a08392 100644

--- a/crmd/crmd_lrm.h

+++ b/crmd/crmd_lrm.h

@@ -71,6 +71,12 @@ struct pending_deletion_op_s {

 xmlNode *do_lrm_query_internal(lrm_state_t * lrm_state, gboolean is_replace);

 /*!

+ * \brief Is this the local ipc connection to the lrmd

+ */

+gboolean

+lrm_state_is_local(lrm_state_t *lrm_state);

+

+/*!

  * \brief Clear all state information from a single state entry.

  * \note This does not close the lrmd connection

  */

diff --git a/crmd/lrm.c b/crmd/lrm.c

index b68f657..5bff4d5 100644

--- a/crmd/lrm.c

+++ b/crmd/lrm.c

@@ -1163,7 +1163,12 @@ get_lrm_resource(lrm_state_t * lrm_state, xmlNode * resource, xmlNode * op_msg,

             fsa_data_t *msg_data = NULL;

             crm_err("Could not add resource %s to LRM %s", id, lrm_state->node_name);

-            register_fsa_error(C_FSA_INTERNAL, I_FAIL, NULL);

+            /* only register this as a internal error if this involves the local

+             * lrmd. Otherwise we're likely dealing with an unresponsive remote-node

+             * which is not a FSA failure. */

+            if (lrm_state_is_local(lrm_state) == TRUE) {

+                register_fsa_error(C_FSA_INTERNAL, I_FAIL, NULL);

+            }

         }

     }

@@ -1211,6 +1216,28 @@ delete_resource(lrm_state_t * lrm_state,

     delete_rsc_entry(lrm_state, request, id, gIter, rc, user);

 }

+static int

+get_fake_call_id(lrm_state_t *lrm_state, const char *rsc_id)

+{

+    int call_id = 0;

+    rsc_history_t *entry = NULL;

+

+    entry = g_hash_table_lookup(lrm_state->resource_history, rsc_id);

+

+    /* Make sure the call id is greater than the last successful operation,

+     * otherwise the failure will not result in a possible recovery of the resource

+     * as it could appear the failure occurred before the successful start */

+    if (entry) {

+        call_id = entry->last_callid + 1;

+    }

+

+    if (call_id < 0) {

+        call_id = 1;

+    }

+    return call_id;

+}

+

+

 /*	 A_LRM_INVOKE	*/

 void

 do_lrm_invoke(long long action,

@@ -1272,7 +1299,6 @@ do_lrm_invoke(long long action,

         operation = CRM_OP_LRM_REFRESH;

     } else if (safe_str_eq(crm_op, CRM_OP_LRM_FAIL)) {

-        rsc_history_t *entry = NULL;

         lrmd_event_data_t *op = NULL;

         lrmd_rsc_info_t *rsc = NULL;

         xmlNode *xml_rsc = find_xml_node(input->xml, XML_CIB_TAG_RESOURCE, TRUE);

@@ -1293,16 +1319,7 @@ do_lrm_invoke(long long action,

         free((char *)op->user_data);

         op->user_data = NULL;

-        entry = g_hash_table_lookup(lrm_state->resource_history, op->rsc_id);

-        /* Make sure the call id is greater than the last successful operation,

-         * otherwise the failure will not result in a possible recovery of the resource

-         * as it could appear the failure occurred before the successful start */

-        if (entry) {

-            op->call_id = entry->last_callid + 1;

-            if (op->call_id < 0) {

-                op->call_id = 1;

-            }

-        }

+        op->call_id = get_fake_call_id(lrm_state, op->rsc_id);

         op->interval = 0;

         op->op_status = PCMK_LRM_OP_DONE;

         op->rc = PCMK_OCF_UNKNOWN_ERROR;

@@ -1430,6 +1447,21 @@ do_lrm_invoke(long long action,

         rsc = get_lrm_resource(lrm_state, xml_rsc, input->xml, create_rsc);

         if (rsc == NULL && create_rsc) {

+            lrmd_event_data_t *op = NULL;

+

+            /* if the operation couldn't complete because we can't register

+             * the resource, return a generic error */

+            op = construct_op(lrm_state, input->xml, ID(xml_rsc), operation);

+            CRM_ASSERT(op != NULL);

+

+            op->op_status = PCMK_LRM_OP_DONE;

+            op->rc = PCMK_OCF_UNKNOWN_ERROR;

+            op->t_run = time(NULL);

+            op->t_rcchange = op->t_run;

+

+            send_direct_ack(from_host, from_sys, NULL, op, ID(xml_rsc));

+            lrmd_free_event(op);

+

             crm_err("Invalid resource definition");

             crm_log_xml_warn(input->msg, "bad input");

@@ -1767,6 +1799,7 @@ do_lrm_rsc_op(lrm_state_t * lrm_state, lrmd_rsc_info_t * rsc, const char *operat

     lrmd_key_value_t *params = NULL;

     fsa_data_t *msg_data = NULL;

     const char *transition = NULL;

+    gboolean stop_recurring = FALSE;

     CRM_CHECK(rsc != NULL, return);

     CRM_CHECK(operation != NULL, return);

@@ -1781,10 +1814,25 @@ do_lrm_rsc_op(lrm_state_t * lrm_state, lrmd_rsc_info_t * rsc, const char *operat

     op = construct_op(lrm_state, msg, rsc->id, operation);

     CRM_CHECK(op != NULL, return);

-    /* stop any previous monitor operations before changing the resource state */

-    if (op->interval == 0

+    if (is_remote_lrmd_ra(NULL, NULL, rsc->id)

+        && op->interval == 0

+        && strcmp(operation, CRMD_ACTION_MIGRATE) == 0) {

+

+        /* pcmk remote connections are a special use case.

+         * We never ever want to stop monitoring a connection resource until

+         * the entire migration has completed. If the connection is ever unexpected

+         * severed, even during a migration, this is an event we must detect.*/

+        stop_recurring = FALSE;

+

+    } else if (op->interval == 0

         && strcmp(operation, CRMD_ACTION_STATUS) != 0

         && strcmp(operation, CRMD_ACTION_NOTIFY) != 0) {

+

+        /* stop any previous monitor operations before changing the resource state */

+        stop_recurring = TRUE;

+    }

+

+    if (stop_recurring == TRUE) {

         guint removed = 0;

         struct stop_recurring_action_s data;

@@ -1837,10 +1885,20 @@ do_lrm_rsc_op(lrm_state_t * lrm_state, lrmd_rsc_info_t * rsc, const char *operat

                              op->op_type,

                              op->user_data, op->interval, op->timeout, op->start_delay, params);

-    if (call_id <= 0) {

+    if (call_id <= 0 && lrm_state_is_local(lrm_state)) {

         crm_err("Operation %s on %s failed: %d", operation, rsc->id, call_id);

         register_fsa_error(C_FSA_INTERNAL, I_FAIL, NULL);

+    } else if (call_id <= 0) {

+

+        crm_err("Operation %s on resource %s failed to execute on remote node %s: %d", operation, rsc->id, lrm_state->node_name, call_id);

+        op->call_id = get_fake_call_id(lrm_state, rsc->id);

+        op->op_status = PCMK_LRM_OP_DONE;

+        op->rc = PCMK_OCF_UNKNOWN_ERROR;

+        op->t_run = time(NULL);

+        op->t_rcchange = op->t_run;

+        process_lrm_event(lrm_state, op);

+

     } else {

         /* record all operations so we can wait

          * for them to complete during shutdown

diff --git a/crmd/lrm_state.c b/crmd/lrm_state.c

index 72f66c7..efd061c 100644

--- a/crmd/lrm_state.c

+++ b/crmd/lrm_state.c

@@ -90,6 +90,21 @@ free_recurring_op(gpointer value)

     free(op);

 }

+gboolean

+lrm_state_is_local(lrm_state_t *lrm_state)

+{

+    if (lrm_state == NULL || fsa_our_uname == NULL) {

+        return FALSE;

+    }

+    

+    if (strcmp(lrm_state->node_name, fsa_our_uname) != 0) {

+        return FALSE;

+    }

+

+    return TRUE;

+

+}

+

 lrm_state_t *

 lrm_state_create(const char *node_name)

 {

diff --git a/crmd/remote_lrmd_ra.c b/crmd/remote_lrmd_ra.c

index 2f658ee..ae59070 100644

--- a/crmd/remote_lrmd_ra.c

+++ b/crmd/remote_lrmd_ra.c

@@ -231,7 +231,7 @@ retry_start_cmd_cb(gpointer data)

         return FALSE;

     }

     cmd = ra_data->cur_cmd;

-    if (safe_str_neq(cmd->action, "start")) {

+    if (safe_str_neq(cmd->action, "start") && safe_str_neq(cmd->action, "migrate_from")) {

         return FALSE;

     }

     update_remaining_timeout(cmd);

@@ -264,7 +264,7 @@ connection_takeover_timeout_cb(gpointer data)

     lrm_state_t *lrm_state = NULL;

     remote_ra_cmd_t *cmd = data;

-    crm_debug("takeover event timed out for node %s", cmd->rsc_id);

+    crm_info("takeover event timed out for node %s", cmd->rsc_id);

     cmd->takeover_timeout_id = 0;

     lrm_state = lrm_state_find(cmd->rsc_id);

@@ -281,7 +281,7 @@ monitor_timeout_cb(gpointer data)

     lrm_state_t *lrm_state = NULL;

     remote_ra_cmd_t *cmd = data;

-    crm_debug("Poke async response timed out for node %s", cmd->rsc_id);

+    crm_info("Poke async response timed out for node %s", cmd->rsc_id);

     cmd->monitor_timeout_id = 0;

     cmd->op_status = PCMK_LRM_OP_TIMEOUT;

     cmd->rc = PCMK_OCF_UNKNOWN_ERROR;

diff --git a/include/crm/pengine/common.h b/include/crm/pengine/common.h

index 33acc6e..0d853c0 100644

--- a/include/crm/pengine/common.h

+++ b/include/crm/pengine/common.h

@@ -34,7 +34,18 @@ enum action_fail_response {

     action_fail_stop,

     action_fail_standby,

     action_fail_fence,

-    action_fail_restart_container

+    action_fail_restart_container,

+    /* This is reserved for internal use for baremetal remote node connection

+     * resources. This fail action means

+     * 1. If stonith is enabled, fence the baremetal remote node

+     * 2. stonith not enabled, attempt to recover the connection resources

+     *

+     * This response value gives us control of saying types of

+     * connection resource failures result in fencing the remote node.

+     * Example: recurring monitors failure should result in fencing.

+     */

+    action_fail_reset_remote,

+

 };

 /* the "done" action must be the "pre" action +1 */

diff --git a/include/crm/pengine/status.h b/include/crm/pengine/status.h

index f08a910..da20f64 100644

--- a/include/crm/pengine/status.h

+++ b/include/crm/pengine/status.h

@@ -138,6 +138,8 @@ struct node_shared_s {

     gboolean is_dc;

     gboolean rsc_discovery_enabled;

+    gboolean remote_requires_reset;

+

     int num_resources;

     GListPtr running_rsc;       /* resource_t* */

     GListPtr allocated_rsc;     /* resource_t* */

diff --git a/lib/common/ipc.c b/lib/common/ipc.c

index f26225f..1b8558d 100644

--- a/lib/common/ipc.c

+++ b/lib/common/ipc.c

@@ -1144,7 +1144,7 @@ crm_ipc_send(crm_ipc_t * client, xmlNode * message, enum crm_ipc_flags flags, in

     id++;

     CRM_LOG_ASSERT(id != 0); /* Crude wrap-around detection */

-    rc = crm_ipc_prepare(id, message, &iov, ipc_buffer_max);

+    rc = crm_ipc_prepare(id, message, &iov, client->max_buf_size);

     if(rc < 0) {

         return rc;

     }

@@ -1158,10 +1158,10 @@ crm_ipc_send(crm_ipc_t * client, xmlNode * message, enum crm_ipc_flags flags, in

     }

     if(header->size_compressed) {

-        if(factor < 10 && (ipc_buffer_max / 10) < (rc / factor)) {

+        if(factor < 10 && (client->max_buf_size / 10) < (rc / factor)) {

             crm_notice("Compressed message exceeds %d0%% of the configured ipc limit (%d bytes), "

                        "consider setting PCMK_ipc_buffer to %d or higher",

-                       factor, ipc_buffer_max, 2*ipc_buffer_max);

+                       factor, client->max_buf_size, 2*client->max_buf_size);

             factor++;

         }

     }

diff --git a/lib/lrmd/lrmd_client.c b/lib/lrmd/lrmd_client.c

index b8c5d23..74bd692 100644

--- a/lib/lrmd/lrmd_client.c

+++ b/lib/lrmd/lrmd_client.c

@@ -604,12 +604,18 @@ lrmd_tls_recv_reply(lrmd_t * lrmd, int total_timeout, int expected_reply_id, int

                 remaining_timeout = total_timeout;

             }

             if (remaining_timeout <= 0) {

+                crm_err("Never received the expected reply during the timeout period, disconnecting.");

+                *disconnected = TRUE;

                 return NULL;

             }

             crm_remote_recv(native->remote, remaining_timeout, disconnected);

             xml = crm_remote_parse_buffer(native->remote);

-            if (!xml || *disconnected) {

+            if (!xml) {

+                crm_err("Unable to receive expected reply, disconnecting.");

+                *disconnected = TRUE;

+                return NULL;

+            } else if (*disconnected) {

                 return NULL;

             }

         }

diff --git a/lib/pengine/common.c b/lib/pengine/common.c

index e5e4ea7..42eda33 100644

--- a/lib/pengine/common.c

+++ b/lib/pengine/common.c

@@ -214,6 +214,9 @@ fail2text(enum action_fail_response fail)

         case action_fail_restart_container:

             result = "restart-container";

             break;

+        case action_fail_reset_remote:

+            result = "reset-remote";

+            break;

     }

     return result;

 }

diff --git a/lib/pengine/unpack.c b/lib/pengine/unpack.c

index a19bdff..bdd6450 100644

--- a/lib/pengine/unpack.c

+++ b/lib/pengine/unpack.c

@@ -80,6 +80,15 @@ pe_fence_node(pe_working_set_t * data_set, node_t * node, const char *reason)

                   node->details->uname, reason);

         set_bit(node->details->remote_rsc->flags, pe_rsc_failed);

+    } else if (is_baremetal_remote_node(node)) {

+        if(pe_can_fence(data_set, node)) {

+            crm_warn("Node %s will be fenced %s", node->details->uname, reason);

+        } else {

+            crm_warn("Node %s is unclean %s", node->details->uname, reason);

+        }

+        node->details->unclean = TRUE;

+        node->details->remote_requires_reset = TRUE;

+

     } else if (node->details->unclean == FALSE) {

         if(pe_can_fence(data_set, node)) {

             crm_warn("Node %s will be fenced %s", node->details->uname, reason);

@@ -1143,8 +1152,10 @@ unpack_remote_status(xmlNode * status, pe_working_set_t * data_set)

         }

         crm_trace("Processing remote node id=%s, uname=%s", id, uname);

-        this_node->details->unclean = FALSE;

-        this_node->details->unseen = FALSE;

+        if (this_node->details->remote_requires_reset == FALSE) {

+            this_node->details->unclean = FALSE;

+            this_node->details->unseen = FALSE;

+        }

         attrs = find_xml_node(state, XML_TAG_TRANSIENT_NODEATTRS, FALSE);

         add_node_attrs(attrs, this_node, TRUE, data_set);

@@ -1777,6 +1788,7 @@ process_rsc_state(resource_t * rsc, node_t * node,

                   enum action_fail_response on_fail,

                   xmlNode * migrate_op, pe_working_set_t * data_set)

 {

+    node_t *tmpnode = NULL;

     CRM_ASSERT(rsc);

     pe_rsc_trace(rsc, "Resource %s is %s on %s: on_fail=%s",

                  rsc->id, role2text(rsc->role), node->details->uname, fail2text(on_fail));

@@ -1817,7 +1829,7 @@ process_rsc_state(resource_t * rsc, node_t * node,

             should_fence = TRUE;

         } else if (is_set(data_set->flags, pe_flag_stonith_enabled)) {

             if (is_baremetal_remote_node(node) && is_not_set(node->details->remote_rsc->flags, pe_rsc_failed)) {

-                /* setting unceen = true means that fencing of the remote node will

+                /* setting unseen = true means that fencing of the remote node will

                  * only occur if the connection resource is not going to start somewhere.

                  * This allows connection resources on a failed cluster-node to move to

                  * another node without requiring the baremetal remote nodes to be fenced

@@ -1892,7 +1904,20 @@ process_rsc_state(resource_t * rsc, node_t * node,

             if (rsc->container) {

                 stop_action(rsc->container, node, FALSE);

-

+            } else if (rsc->role != RSC_ROLE_STOPPED && rsc->role != RSC_ROLE_UNKNOWN) {

+                stop_action(rsc, node, FALSE);

+            }

+            break;

+        case action_fail_reset_remote:

+            set_bit(rsc->flags, pe_rsc_failed);

+            tmpnode = NULL;

+            if (rsc->is_remote_node) {

+                tmpnode = pe_find_node(data_set->nodes, rsc->id);

+            }

+            if (tmpnode && is_baremetal_remote_node(tmpnode)) {

+                /* connection resource to baremetal resource failed in a way that

+                 * should result in fencing the remote-node. */

+                pe_fence_node(data_set, tmpnode, "because of connection failure(s)");

             } else if (rsc->role != RSC_ROLE_STOPPED && rsc->role != RSC_ROLE_UNKNOWN) {

                 stop_action(rsc, node, FALSE);

             }

@@ -1904,7 +1929,7 @@ process_rsc_state(resource_t * rsc, node_t * node,

      * result in a fencing operation regardless if we're going to attempt to 

      * reconnect to the remote-node in this transition or not. */

     if (is_set(rsc->flags, pe_rsc_failed) && rsc->is_remote_node) {

-        node_t *tmpnode = pe_find_node(data_set->nodes, rsc->id);

+        tmpnode = pe_find_node(data_set->nodes, rsc->id);

         if (tmpnode && tmpnode->details->unclean) {

             tmpnode->details->unseen = FALSE;

         }

@@ -2510,10 +2535,9 @@ unpack_rsc_op_failure(resource_t *rsc, node_t *node, int rc, xmlNode *xml_op, en

     action = custom_action(rsc, strdup(key), task, NULL, TRUE, FALSE, data_set);

     if ((action->on_fail <= action_fail_fence && *on_fail < action->on_fail) ||

-        (action->on_fail == action_fail_restart_container

-         && *on_fail <= action_fail_recover) || (*on_fail == action_fail_restart_container

-                                                 && action->on_fail >=

-                                                 action_fail_migrate)) {

+        (action->on_fail == action_fail_reset_remote && *on_fail <= action_fail_recover) ||

+        (action->on_fail == action_fail_restart_container && *on_fail <= action_fail_recover) ||

+        (*on_fail == action_fail_restart_container && action->on_fail >= action_fail_migrate)) {

         pe_rsc_trace(rsc, "on-fail %s -> %s for %s (%s)", fail2text(*on_fail),

                      fail2text(action->on_fail), action->uuid, key);

         *on_fail = action->on_fail;

@@ -2881,13 +2905,11 @@ update_resource_state(resource_t *rsc, node_t * node, xmlNode * xml_op, const ch

             case action_fail_block:

             case action_fail_ignore:

             case action_fail_recover:

-                *on_fail = action_fail_ignore;

-                rsc->next_role = RSC_ROLE_UNKNOWN;

-                break;

-

             case action_fail_restart_container:

+            case action_fail_reset_remote:

                 *on_fail = action_fail_ignore;

                 rsc->next_role = RSC_ROLE_UNKNOWN;

+                break;

         }

     }

 }

diff --git a/lib/pengine/utils.c b/lib/pengine/utils.c

index baf6cb9..a40de60 100644

--- a/lib/pengine/utils.c

+++ b/lib/pengine/utils.c

@@ -35,6 +35,7 @@ void unpack_operation(action_t * action, xmlNode * xml_obj, resource_t * contain

                       pe_working_set_t * data_set);

 static xmlNode *find_rsc_op_entry_helper(resource_t * rsc, const char *key,

                                          gboolean include_disabled);

+static gboolean is_rsc_baremetal_remote_node(resource_t *rsc, pe_working_set_t * data_set);

 bool pe_can_fence(pe_working_set_t * data_set, node_t *node)

 {

@@ -686,6 +687,19 @@ unpack_operation(action_t * action, xmlNode * xml_obj, resource_t * container,

                                NULL, action->meta, NULL, FALSE, data_set->now);

     g_hash_table_remove(action->meta, "id");

+    field = XML_LRM_ATTR_INTERVAL;

+    value = g_hash_table_lookup(action->meta, field);

+    if (value != NULL) {

+        interval = crm_get_interval(value);

+        if (interval > 0) {

+            value_ms = crm_itoa(interval);

+            g_hash_table_replace(action->meta, strdup(field), value_ms);

+

+        } else {

+            g_hash_table_remove(action->meta, field);

+        }

+    }

+

     /* Begin compatability code */

     value = g_hash_table_lookup(action->meta, "requires");

@@ -789,6 +803,22 @@ unpack_operation(action_t * action, xmlNode * xml_obj, resource_t * container,

         action->on_fail = action_fail_restart_container;

         value = "restart container (and possibly migrate) (default)";

+    /* for baremetal remote nodes, ensure that a recurring monitor operation failure

+     * defaults to either fencing the remote-node for recovery, or at least

+     * attempting to recover the the connection when fencing is disabled. */

+    } else if (value == NULL &&

+               is_rsc_baremetal_remote_node(action->rsc, data_set) &&

+               safe_str_eq(action->task, CRMD_ACTION_STATUS) &&

+               interval > 0) {

+

+        if (is_set(data_set->flags, pe_flag_stonith_enabled)) {

+            action->on_fail = action_fail_reset_remote;

+            value = "fence baremetal remote node (default)";

+        } else {

+            action->on_fail = action_fail_recover;

+            value = "recover baremetal remote node connection (default)";

+        }

+

     } else if (value == NULL && safe_str_eq(action->task, CRMD_ACTION_STOP)) {

         if (is_set(data_set->flags, pe_flag_stonith_enabled)) {

             action->on_fail = action_fail_fence;

@@ -824,19 +854,6 @@ unpack_operation(action_t * action, xmlNode * xml_obj, resource_t * container,

     pe_rsc_trace(action->rsc, "\t%s failure results in: %s", action->task,

                  role2text(action->fail_role));

-    field = XML_LRM_ATTR_INTERVAL;

-    value = g_hash_table_lookup(action->meta, field);

-    if (value != NULL) {

-        interval = crm_get_interval(value);

-        if (interval > 0) {

-            value_ms = crm_itoa(interval);

-            g_hash_table_replace(action->meta, strdup(field), value_ms);

-

-        } else {

-            g_hash_table_remove(action->meta, field);

-        }

-    }

-

     field = XML_OP_ATTR_START_DELAY;

     value = g_hash_table_lookup(action->meta, field);

     if (value != NULL) {

@@ -1939,6 +1956,25 @@ const char *rsc_printable_id(resource_t *rsc)

 }

 gboolean

+is_rsc_baremetal_remote_node(resource_t *rsc, pe_working_set_t * data_set)

+{

+    node_t *node;

+

+    if (rsc == NULL) {

+        return FALSE;

+    } else if (rsc->is_remote_node == FALSE) {

+        return FALSE;

+    }

+

+    node = pe_find_node(data_set->nodes, rsc->id);

+    if (node == NULL) {

+        return FALSE;

+    }

+

+    return is_baremetal_remote_node(node);

+}

+

+gboolean

 is_baremetal_remote_node(node_t *node)

 {

     if (is_remote_node(node) && (node->details->remote_rsc == FALSE || node->details->remote_rsc->container == FALSE)) {

diff --git a/pengine/allocate.c b/pengine/allocate.c

index 45e2212..9c3cb60 100644

--- a/pengine/allocate.c

+++ b/pengine/allocate.c

@@ -1191,7 +1191,10 @@ allocate_resources(pe_working_set_t * data_set)

                 continue;

             }

             pe_rsc_trace(rsc, "Allocating: %s", rsc->id);

-            rsc->cmds->allocate(rsc, NULL, data_set);

+            /* for remote node connection resources, always prefer the partial migration

+             * target during resource allocation if the rsc is in the middle of a

+             * migration */ 

+            rsc->cmds->allocate(rsc, rsc->partial_migration_target, data_set);

         }

     }

@@ -1716,9 +1719,18 @@ apply_remote_node_ordering(pe_working_set_t *data_set)

                 node_t *cluster_node = remote_rsc->running_on ? remote_rsc->running_on->data : NULL;

                 /* if the current cluster node a baremetal connection resource

-                 * is residing on is unclean, we can't process any operations on that

-                 * remote node until after it starts somewhere else. */

-                if (cluster_node && cluster_node->details->unclean == TRUE) {

+                 * is residing on is unclean or went offline we can't process any

+                 * operations on that remote node until after it starts somewhere else. */

+                if (cluster_node == NULL ||

+                    cluster_node->details->unclean == TRUE ||

+                    cluster_node->details->online == FALSE) {

+                    after_start = TRUE;

+                } else if (g_list_length(remote_rsc->running_on) > 1 &&

+                           remote_rsc->partial_migration_source &&

+                            remote_rsc->partial_migration_target) {

+                    /* if we're caught in the middle of migrating a connection resource,

+                     * then we have to wait until after the resource migrates before performing

+                     * any actions. */

                     after_start = TRUE;

                 }

             }

diff --git a/pengine/native.c b/pengine/native.c

index 6d62010..62639d0 100644

--- a/pengine/native.c

+++ b/pengine/native.c

@@ -1137,12 +1137,18 @@ handle_migration_actions(resource_t * rsc, node_t *current, node_t *chosen, pe_w

     if (migrate_to) {

         add_hash_param(migrate_to->meta, XML_LRM_ATTR_MIGRATE_SOURCE, current->details->uname);

         add_hash_param(migrate_to->meta, XML_LRM_ATTR_MIGRATE_TARGET, chosen->details->uname);

-        /* migrate_to takes place on the source node, but can 

-         * have an effect on the target node depending on how

-         * the agent is written. Because of this, we have to maintain

-         * a record that the migrate_to occurred incase the source node 

-         * loses membership while the migrate_to action is still in-flight. */

-        add_hash_param(migrate_to->meta, XML_OP_ATTR_PENDING, "true");

+

+        /* pcmk remote connections don't require pending to be recorded in cib.

+         * We can optimize cib writes by only setting PENDING for non pcmk remote

+         * connection resources */

+        if (rsc->is_remote_node == FALSE) {

+            /* migrate_to takes place on the source node, but can 

+             * have an effect on the target node depending on how

+             * the agent is written. Because of this, we have to maintain

+             * a record that the migrate_to occurred incase the source node 

+             * loses membership while the migrate_to action is still in-flight. */

+            add_hash_param(migrate_to->meta, XML_OP_ATTR_PENDING, "true");

+        }

     }

     if (migrate_from) {

diff --git a/pengine/regression.sh b/pengine/regression.sh

index aa03b18..3a244d7 100755

--- a/pengine/regression.sh

+++ b/pengine/regression.sh

@@ -769,6 +769,11 @@ do_test remote-disable         "Disable a baremetal remote-node"

 do_test remote-orphaned        "Properly shutdown orphaned connection resource"

 do_test remote-recover         "Recover connection resource after cluster-node fails."

 do_test remote-stale-node-entry "Make sure we properly handle leftover remote-node entries in the node section"

+do_test remote-partial-migrate  "Make sure partial migrations are handled before ops on the remote node."

+do_test remote-partial-migrate2 "Make sure partial migration target is prefered for remote connection."

+do_test remote-recover-fail     "Make sure start failure causes fencing if rsc are active on remote."

+do_test remote-start-fail       "Make sure a start failure does not result in fencing if no active resources are on remote."

+do_test remote-unclean2         "Make monitor failure always results in fencing, even if no rsc are active on remote."

 echo ""

 do_test resource-discovery      "Exercises resource-discovery location constraint option."

diff --git a/pengine/test10/remote-fence-unclean.exp b/pengine/test10/remote-fence-unclean.exp

index 9b1e327..b8c90f2 100644

--- a/pengine/test10/remote-fence-unclean.exp

+++ b/pengine/test10/remote-fence-unclean.exp

@@ -8,7 +8,7 @@

     </action_set>

     <inputs>

       <trigger>

-        <rsc_op id="5" operation="stop" operation_key="remote1_stop_0" on_node="18node1" on_node_uuid="1"/>

+        <rsc_op id="14" operation="stop" operation_key="remote1_stop_0" on_node="18node1" on_node_uuid="1"/>

       </trigger>

       <trigger>

         <pseudo_event id="27" operation="stonith_complete" operation_key="stonith_complete"/>

@@ -17,7 +17,7 @@

   </synapse>

   <synapse id="1">

     <action_set>

-      <rsc_op id="5" operation="stop" operation_key="remote1_stop_0" on_node="18node1" on_node_uuid="1">

+      <rsc_op id="14" operation="stop" operation_key="remote1_stop_0" on_node="18node1" on_node_uuid="1">

         <primitive id="remote1" class="ocf" provider="pacemaker" type="remote"/>

         <attributes CRM_meta_timeout="20000" />

       </rsc_op>

@@ -176,13 +176,13 @@

   </synapse>

   <synapse id="14">

     <action_set>

-      <pseudo_event id="7" operation="all_stopped" operation_key="all_stopped">

+      <pseudo_event id="6" operation="all_stopped" operation_key="all_stopped">

         <attributes />

       </pseudo_event>

     </action_set>

     <inputs>

       <trigger>

-        <rsc_op id="5" operation="stop" operation_key="remote1_stop_0" on_node="18node1" on_node_uuid="1"/>

+        <rsc_op id="14" operation="stop" operation_key="remote1_stop_0" on_node="18node1" on_node_uuid="1"/>

       </trigger>

       <trigger>

         <pseudo_event id="18" operation="stop" operation_key="FAKE2_stop_0"/>

diff --git a/pengine/test10/remote-move.exp b/pengine/test10/remote-move.exp

index 6724679..6287daa 100644

--- a/pengine/test10/remote-move.exp

+++ b/pengine/test10/remote-move.exp

@@ -51,7 +51,7 @@

     <action_set>

       <rsc_op id="20" operation="migrate_to" operation_key="remote1_migrate_to_0" on_node="18builder" on_node_uuid="5">

         <primitive id="remote1" class="ocf" provider="pacemaker" type="remote"/>

-        <attributes CRM_meta_migrate_source="18builder" CRM_meta_migrate_target="18node1" CRM_meta_record_pending="true" CRM_meta_timeout="20000" />

+        <attributes CRM_meta_migrate_source="18builder" CRM_meta_migrate_target="18node1" CRM_meta_timeout="20000" />

       </rsc_op>

     </action_set>

     <inputs/>

diff --git a/pengine/test10/remote-partial-migrate.dot b/pengine/test10/remote-partial-migrate.dot

new file mode 100644

index 0000000..8d776dc

--- /dev/null

+++ b/pengine/test10/remote-partial-migrate.dot

@@ -0,0 +1,86 @@

+ digraph "g" {

+"FAKE10_monitor_10000 pcmk1" [ style=bold color="green" fontcolor="black"]

+"FAKE10_start_0 pcmk1" -> "FAKE10_monitor_10000 pcmk1" [ style = bold]

+"FAKE10_start_0 pcmk1" [ style=bold color="green" fontcolor="black"]

+"FAKE13_monitor_10000 pcmk2" [ style=bold color="green" fontcolor="black"]

+"FAKE13_start_0 pcmk2" -> "FAKE13_monitor_10000 pcmk2" [ style = bold]

+"FAKE13_start_0 pcmk2" [ style=bold color="green" fontcolor="black"]

+"FAKE15_monitor_10000 pcmk3" [ style=bold color="green" fontcolor="black"]

+"FAKE15_start_0 pcmk3" -> "FAKE15_monitor_10000 pcmk3" [ style = bold]

+"FAKE15_start_0 pcmk3" [ style=bold color="green" fontcolor="black"]

+"FAKE16_monitor_10000 pcmk_remote3" [ style=bold color="green" fontcolor="black"]

+"FAKE16_start_0 pcmk_remote3" -> "FAKE16_monitor_10000 pcmk_remote3" [ style = bold]

+"FAKE16_start_0 pcmk_remote3" [ style=bold color="green" fontcolor="black"]

+"FAKE16_stop_0 pcmk1" -> "FAKE16_start_0 pcmk_remote3" [ style = bold]

+"FAKE16_stop_0 pcmk1" -> "all_stopped" [ style = bold]

+"FAKE16_stop_0 pcmk1" [ style=bold color="green" fontcolor="black"]

+"FAKE22_monitor_10000 pcmk1" [ style=bold color="green" fontcolor="black"]

+"FAKE22_start_0 pcmk1" -> "FAKE22_monitor_10000 pcmk1" [ style = bold]

+"FAKE22_start_0 pcmk1" [ style=bold color="green" fontcolor="black"]

+"FAKE23_monitor_10000 pcmk_remote1" [ style=bold color="green" fontcolor="black"]

+"FAKE23_start_0 pcmk_remote1" -> "FAKE23_monitor_10000 pcmk_remote1" [ style = bold]

+"FAKE23_start_0 pcmk_remote1" [ style=bold color="green" fontcolor="black"]

+"FAKE23_stop_0 pcmk1" -> "FAKE23_start_0 pcmk_remote1" [ style = bold]

+"FAKE23_stop_0 pcmk1" -> "all_stopped" [ style = bold]

+"FAKE23_stop_0 pcmk1" [ style=bold color="green" fontcolor="black"]

+"FAKE26_monitor_10000 pcmk1" [ style=bold color="green" fontcolor="black"]

+"FAKE26_start_0 pcmk1" -> "FAKE26_monitor_10000 pcmk1" [ style = bold]

+"FAKE26_start_0 pcmk1" [ style=bold color="green" fontcolor="black"]

+"FAKE29_monitor_10000 pcmk2" [ style=bold color="green" fontcolor="black"]

+"FAKE29_start_0 pcmk2" -> "FAKE29_monitor_10000 pcmk2" [ style = bold]

+"FAKE29_start_0 pcmk2" [ style=bold color="green" fontcolor="black"]

+"FAKE30_monitor_10000 pcmk_remote2" [ style=bold color="green" fontcolor="black"]

+"FAKE30_start_0 pcmk_remote2" -> "FAKE30_monitor_10000 pcmk_remote2" [ style = bold]

+"FAKE30_start_0 pcmk_remote2" [ style=bold color="green" fontcolor="black"]

+"FAKE30_stop_0 pcmk1" -> "FAKE30_start_0 pcmk_remote2" [ style = bold]

+"FAKE30_stop_0 pcmk1" -> "all_stopped" [ style = bold]

+"FAKE30_stop_0 pcmk1" [ style=bold color="green" fontcolor="black"]

+"FAKE36_monitor_10000 pcmk1" [ style=bold color="green" fontcolor="black"]

+"FAKE36_start_0 pcmk1" -> "FAKE36_monitor_10000 pcmk1" [ style = bold]