ldap/servers/plugins/replication/repl5.h | 1
ldap/servers/plugins/replication/repl5_connection.c | 19 -
ldap/servers/plugins/replication/repl5_inc_protocol.c | 180 ++++++++++--------
ldap/servers/plugins/replication/repl5_plugins.c | 60 +++++-
ldap/servers/plugins/replication/urp.c | 2
5 files changed, 167 insertions(+), 95 deletions(-)
New commits:
commit 80c68e202ff2b50a60f35b6683ef26b41609bc56
Author: Mark Reynolds <mreynolds(a)redhat.com>
Date: Sun Jan 17 19:33:44 2016 -0500
Ticket 47788 - Supplier can skip a failing update, although it should retry
Bug Description: If a replicated update fails on the consumer,
the update is never tried. This is due to the
replication async result thread missing the failure
before another update is replicated and it succeeds.
This second update that succeeds updates the consumer
RUV. This makes it appear that the consumer is caught
up, and the supplier never resends that original
failed update.
Fix Description: When a replicated update fails, and its an error we can
not ignore, the connection is closed. Which stops the
replication session, and prevents any further updates
coming in and updating the consumer RUV. This allows
the supplier to correctly retry the operation that
failed on the next replication session.
https://fedorahosted.org/389/ticket/47788
Reviewed by: nhosoi, wibrown, and rmeggins (Thanks!!!)
diff --git a/ldap/servers/plugins/replication/repl5.h
b/ldap/servers/plugins/replication/repl5.h
index 7d85631..ac467eb 100644
--- a/ldap/servers/plugins/replication/repl5.h
+++ b/ldap/servers/plugins/replication/repl5.h
@@ -612,6 +612,7 @@ PRUint64 replica_get_backoff_min(Replica *r);
PRUint64 replica_get_backoff_max(Replica *r);
void replica_set_backoff_min(Replica *r, PRUint64 min);
void replica_set_backoff_max(Replica *r, PRUint64 max);
+PRBool ignore_error_and_keep_going(int error);
/* The functions below handles the state flag */
/* Current internal state flags */
diff --git a/ldap/servers/plugins/replication/repl5_connection.c
b/ldap/servers/plugins/replication/repl5_connection.c
index 918c5ea..8f4a6b6 100644
--- a/ldap/servers/plugins/replication/repl5_connection.c
+++ b/ldap/servers/plugins/replication/repl5_connection.c
@@ -473,17 +473,17 @@ conn_read_result_ex(Repl_Connection *conn, char **retoidp, struct
berval **retda
conn->last_ldap_error = rc;
close_connection_internal(conn); /* we already have the lock */
return_value = CONN_NOT_CONNECTED;
+ goto done;
}
else if (IS_DISCONNECT_ERROR(err))
{
conn->last_ldap_error = err;
close_connection_internal(conn); /* we already have the lock */
return_value = CONN_NOT_CONNECTED;
+ goto done;
}
/* Got a result */
- if ((rc == LDAP_SUCCESS) && (err == LDAP_BUSY))
- return_value = CONN_BUSY;
- else if (retoidp)
+ if (retoidp /* total update */)
{
if (!((rc == LDAP_SUCCESS) && (err == LDAP_BUSY)))
{
@@ -512,16 +512,11 @@ conn_read_result_ex(Repl_Connection *conn, char **retoidp, struct
berval **retda
}
return_value = LDAP_SUCCESS == conn->last_ldap_error ? CONN_OPERATION_SUCCESS :
CONN_OPERATION_FAILED;
}
- /*
- * XXXggood do I need to free matched, referrals,
- * anything else? Or can I pass NULL for the args
- * I'm not interested in?
- */
- /* Good question! Meanwhile, as RTM aproaches, let's free them... */
- slapi_ch_free((void **) &errmsg);
- slapi_ch_free((void **) &matched);
- charray_free(referrals);
conn->status = STATUS_CONNECTED;
+done:
+ slapi_ch_free_string(&errmsg);
+ slapi_ch_free_string(&matched);
+ charray_free(referrals);
}
if (res) ldap_msgfree(res);
PR_Unlock(conn->lock); /* release the conn lock */
diff --git a/ldap/servers/plugins/replication/repl5_inc_protocol.c
b/ldap/servers/plugins/replication/repl5_inc_protocol.c
index 5cf170c..e02e883 100644
--- a/ldap/servers/plugins/replication/repl5_inc_protocol.c
+++ b/ldap/servers/plugins/replication/repl5_inc_protocol.c
@@ -174,7 +174,6 @@ static void protocol_sleep(Private_Repl_Protocol *prp, PRIntervalTime
duration);
static int send_updates(Private_Repl_Protocol *prp, RUV *ruv, PRUint32
*num_changes_sent);
static void repl5_inc_backoff_expired(time_t timer_fire_time, void *arg);
static int examine_update_vector(Private_Repl_Protocol *prp, RUV *ruv);
-static PRBool ignore_error_and_keep_going(int error);
static const char* state2name (int state);
static const char* event2name (int event);
static const char* op2string (int op);
@@ -478,11 +477,13 @@ repl5_inc_flow_control_results(Repl_Agmt *agmt, result_data *rd)
PR_Unlock(rd->lock);
}
-static void
+static int
repl5_inc_waitfor_async_results(result_data *rd)
{
int done = 0;
int loops = 0;
+ int rc = UPDATE_NO_MORE_UPDATES;
+
/* Keep pulling results off the LDAP connection until we catch up to the last message id
stored in the rd */
while (!done && !slapi_is_shutting_down())
{
@@ -501,6 +502,10 @@ repl5_inc_waitfor_async_results(result_data *rd)
{
done = 1; /* no connection == no more results */
}
+ /*
+ * Return the last operation result
+ */
+ rc = rd->result;
PR_Unlock(rd->lock);
/* If not then sleep a bit */
DS_Sleep(PR_SecondsToInterval(1));
@@ -516,6 +521,7 @@ repl5_inc_waitfor_async_results(result_data *rd)
done = 1;
}
}
+ return rc;
}
/*
@@ -1496,78 +1502,84 @@ static int
repl5_inc_update_from_op_result(Private_Repl_Protocol *prp, ConnResult replay_crc, int
connection_error, char *csn_str, char *uniqueid, ReplicaId replica_id, int* finished,
PRUint32 *num_changes_sent)
{
int return_value = 0;
-
- /* Indentation is wrong here so we can get a sensible cvs diff */
- if (CONN_OPERATION_SUCCESS != replay_crc)
- {
- /* Figure out what to do next */
- if (CONN_OPERATION_FAILED == replay_crc)
- {
- /* Map ldap error code to return value */
- if (!ignore_error_and_keep_going(connection_error))
- {
- return_value = UPDATE_TRANSIENT_ERROR;
- *finished = 1;
- }
- else
- {
- agmt_inc_last_update_changecount (prp->agmt, replica_id, 1 /*skipped*/);
- }
- slapi_log_error(*finished ? SLAPI_LOG_FATAL : slapi_log_urp, repl_plugin_name,
- "%s: Consumer failed to replay change (uniqueid %s, CSN %s): %s (%d).
%s.\n",
- agmt_get_long_name(prp->agmt),
- uniqueid, csn_str,
- ldap_err2string(connection_error), connection_error,
- *finished ? "Will retry later" : "Skipping");
- }
- else if (CONN_NOT_CONNECTED == replay_crc)
- {
- /* We lost the connection - enter backoff state */
- return_value = UPDATE_CONNECTION_LOST;
- *finished = 1;
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
- "%s: Consumer failed to replay change (uniqueid %s, CSN %s): "
- "%s(%d). Will retry later.\n",
- agmt_get_long_name(prp->agmt),
- uniqueid, csn_str,
- connection_error ? ldap_err2string(connection_error) : "Connection
lost",
- connection_error);
- }
- else if (CONN_TIMEOUT == replay_crc)
- {
- return_value = UPDATE_TIMEOUT;
- *finished = 1;
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
- "%s: Consumer timed out to replay change (uniqueid %s, CSN %s): "
- "%s.\n",
- agmt_get_long_name(prp->agmt),
- uniqueid, csn_str,
- connection_error ? ldap_err2string(connection_error) : "Timeout");
- }
- else if (CONN_LOCAL_ERROR == replay_crc)
- {
- /*
- * Something bad happened on the local server - enter
- * backoff state.
- */
- return_value = UPDATE_TRANSIENT_ERROR;
- *finished = 1;
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
- "%s: Failed to replay change (uniqueid %s, CSN %s): "
- "Local error. Will retry later.\n",
- agmt_get_long_name(prp->agmt),
- uniqueid, csn_str);
- }
-
- }
- else
- {
- /* Positive response received */
- (*num_changes_sent)++;
- agmt_inc_last_update_changecount (prp->agmt, replica_id, 0 /*replayed*/);
- }
- return return_value;
+ if (CONN_OPERATION_SUCCESS != replay_crc)
+ {
+ /* Figure out what to do next */
+ if (CONN_OPERATION_FAILED == replay_crc)
+ {
+ /* Map ldap error code to return value */
+ if (!ignore_error_and_keep_going(connection_error))
+ {
+ return_value = UPDATE_TRANSIENT_ERROR;
+ *finished = 1;
+ }
+ else
+ {
+ agmt_inc_last_update_changecount (prp->agmt, replica_id, 1 /*skipped*/);
+ }
+ slapi_log_error(*finished ? SLAPI_LOG_FATAL : slapi_log_urp, repl_plugin_name,
+ "%s: Consumer failed to replay change (uniqueid %s, CSN %s): %s (%d).
%s.\n",
+ agmt_get_long_name(prp->agmt),
+ uniqueid, csn_str,
+ ldap_err2string(connection_error), connection_error,
+ *finished ? "Will retry later" : "Skipping");
+ }
+ else if (CONN_NOT_CONNECTED == replay_crc)
+ {
+ /* We lost the connection - enter backoff state */
+
+ return_value = UPDATE_CONNECTION_LOST;
+ *finished = 1;
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ "%s: Consumer failed to replay change (uniqueid %s, CSN %s): "
+ "%s(%d). Will retry later.\n",
+ agmt_get_long_name(prp->agmt),
+ uniqueid, csn_str,
+ connection_error ? ldap_err2string(connection_error) : "Connection lost",
+ connection_error);
+ }
+ else if (CONN_TIMEOUT == replay_crc)
+ {
+ return_value = UPDATE_TIMEOUT;
+ *finished = 1;
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ "%s: Consumer timed out to replay change (uniqueid %s, CSN %s): "
+ "%s.\n",
+ agmt_get_long_name(prp->agmt),
+ uniqueid, csn_str,
+ connection_error ? ldap_err2string(connection_error) : "Timeout");
+ }
+ else if (CONN_LOCAL_ERROR == replay_crc)
+ {
+ /*
+ * Something bad happened on the local server - enter
+ * backoff state.
+ */
+ return_value = UPDATE_TRANSIENT_ERROR;
+ *finished = 1;
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ "%s: Failed to replay change (uniqueid %s, CSN %s): "
+ "Local error. Will retry later.\n",
+ agmt_get_long_name(prp->agmt),
+ uniqueid, csn_str);
+ }
+ if (*finished){
+ /*
+ * A serious error has occurred, the consumer might have closed
+ * the connection already, but we need to close the conn on the
+ * supplier side to properly set the conn structure as closed.
+ */
+ conn_disconnect(prp->conn);
+ }
+ }
+ else
+ {
+ /* Positive response received */
+ (*num_changes_sent)++;
+ agmt_inc_last_update_changecount (prp->agmt, replica_id, 0 /*replayed*/);
+ }
+ return return_value;
}
/*
@@ -1585,7 +1597,7 @@ send_updates(Private_Repl_Protocol *prp, RUV *remote_update_vector,
PRUint32 *nu
{
CL5Entry entry;
slapi_operation_parameters op;
- int return_value;
+ int return_value = 0;
int rc;
CL5ReplayIterator *changelog_iterator;
int message_id = 0;
@@ -1912,8 +1924,22 @@ send_updates(Private_Repl_Protocol *prp, RUV *remote_update_vector,
PRUint32 *nu
{
/* We need to ensure that we wait until all the responses have been received from our
operations */
if (return_value != UPDATE_CONNECTION_LOST) {
- /* if connection was lost/closed, there will be nothing to read */
- repl5_inc_waitfor_async_results(rd);
+ /*
+ * If we already have an error, there is no need to check the
+ * async result thread anymore.
+ */
+ if (return_value == UPDATE_NO_MORE_UPDATES)
+ {
+ /*
+ * We need to double check that an error hasn't popped up from
+ * the async result thread since our last check.
+ */
+ int final_result;
+
+ if((final_result = repl5_inc_waitfor_async_results(rd))){
+ return_value = final_result;
+ }
+ }
}
rc = repl5_inc_destroy_async_result_thread(rd);
@@ -2202,7 +2228,7 @@ examine_update_vector(Private_Repl_Protocol *prp, RUV *remote_ruv)
* We stop if there's some indication that the server just completely
* failed to process the operation, e.g. LDAP_OPERATIONS_ERROR.
*/
-static PRBool
+PRBool
ignore_error_and_keep_going(int error)
{
int return_value = PR_FALSE;
diff --git a/ldap/servers/plugins/replication/repl5_plugins.c
b/ldap/servers/plugins/replication/repl5_plugins.c
index 5adcb6c..c2ca2e4 100644
--- a/ldap/servers/plugins/replication/repl5_plugins.c
+++ b/ldap/servers/plugins/replication/repl5_plugins.c
@@ -1208,12 +1208,13 @@ write_changelog_and_ruv (Slapi_PBlock *pb)
static int
process_postop (Slapi_PBlock *pb)
{
- int rc = LDAP_SUCCESS;
- Slapi_Operation *op;
+ Slapi_Operation *op;
Slapi_Backend *be;
- int is_replicated_operation = 0;
+ int is_replicated_operation = 0;
CSN *opcsn = NULL;
char sessionid[REPL_SESSION_ID_SIZE];
+ int retval = LDAP_SUCCESS;
+ int rc = 0;
/* we just let fixup operations through */
slapi_pblock_get( pb, SLAPI_OPERATION, &op );
@@ -1237,8 +1238,8 @@ process_postop (Slapi_PBlock *pb)
get_repl_session_id (pb, sessionid, &opcsn);
- slapi_pblock_get(pb, SLAPI_RESULT_CODE, &rc);
- if (rc == LDAP_SUCCESS)
+ slapi_pblock_get(pb, SLAPI_RESULT_CODE, &retval);
+ if (retval == LDAP_SUCCESS)
{
agmtlist_notify_all(pb);
rc = SLAPI_PLUGIN_SUCCESS;
@@ -1283,6 +1284,55 @@ process_postop (Slapi_PBlock *pb)
slapi_ch_free((void **)
&op_params->p.p_modrdn.modrdn_newsuperior_address.uniqueid);
}
}
+ if (!ignore_error_and_keep_going(retval)){
+ /*
+ * We have an error we can't ignore. Release the replica and close
+ * the connection to stop the replication session.
+ */
+ consumer_connection_extension *connext = NULL;
+ Slapi_Connection *conn = NULL;
+ char csn_str[CSN_STRSIZE] = {'\0'};
+ PRUint64 connid = 0;
+ int opid = 0;
+
+ slapi_pblock_get(pb, SLAPI_CONNECTION, &conn);
+ slapi_pblock_get(pb, SLAPI_OPERATION_ID, &opid);
+ slapi_pblock_get(pb, SLAPI_CONN_ID, &connid);
+ if (conn)
+ {
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ "process_postop: Failed to apply update (%s) error (%d). "
+ "Aborting replication session(conn=%" NSPRIu64 " op=%d)\n",
+ csn_as_string(opcsn, PR_FALSE, csn_str), retval,
+ (long long unsigned int)connid, opid);
+ /*
+ * Release this replica so new sessions can begin
+ */
+ connext = consumer_connection_extension_acquire_exclusive_access(conn, connid, opid);
+ if (connext && connext->replica_acquired)
+ {
+ int zero = 0;
+ Replica *r = (Replica*)object_get_data ((Object*)connext->replica_acquired);
+
+ replica_relinquish_exclusive_access(r, connid, opid);
+ object_release ((Object*)connext->replica_acquired);
+ connext->replica_acquired = NULL;
+ connext->isreplicationsession = 0;
+ slapi_pblock_set( pb, SLAPI_CONN_IS_REPLICATION_SESSION, &zero );
+ }
+ if (connext){
+ consumer_connection_extension_relinquish_exclusive_access(conn, connid, opid,
PR_FALSE);
+ }
+
+ /*
+ * Close the connection to end the current session with the
+ * supplier. This prevents new updates from coming in and
+ * updating the consumer RUV - which would cause this failed
+ * update to be never be replayed.
+ */
+ slapi_disconnect_server(conn);
+ }
+ }
if (NULL == opcsn)
opcsn = operation_get_csn(op);
if (opcsn)
diff --git a/ldap/servers/plugins/replication/urp.c
b/ldap/servers/plugins/replication/urp.c
index c7d60e2..3045e06 100644
--- a/ldap/servers/plugins/replication/urp.c
+++ b/ldap/servers/plugins/replication/urp.c
@@ -151,7 +151,7 @@ urp_add_operation( Slapi_PBlock *pb )
slapi_log_error(slapi_log_urp, sessionid,
"urp_add (%s): an entry with this uniqueid already exists.\n",
slapi_entry_get_dn_const(existing_uniqueid_entry));
- op_result= LDAP_UNWILLING_TO_PERFORM;
+ op_result= LDAP_ALREADY_EXISTS;
slapi_pblock_set(pb, SLAPI_RESULT_CODE, &op_result);
rc = SLAPI_PLUGIN_NOOP; /* Ignore this Operation */
PROFILE_POINT; /* Add Conflict; UniqueID Exists; Ignore */