From 3d53f4c1c9ce5011fa88ab71a4f2742a17e6fdb3 Mon Sep 17 00:00:00 2001
From: Mark Reynolds <mreynolds@redhat.com>
Date: Wed, 8 Jun 2016 13:06:46 -0400
Subject: [PATCH] Ticket 48636 - Improve replication convergence

Bug Description:  In a busy MMR environment where multiple masters are being
                  updated at the same time the replica sessions stay open for
                  a very long time.  This causes other masters to wait to send
                  their updates.  This causes lop-sided convergence.  Where
                  entries added to the MMR environment, but on different masters,
                  take a very different amount of time until they are each seen
                  on all the replicas.

Fix Description:  A new configuratoin setting was added (nsds5ReplicaReleaseTimeout)
                  to the replica configuration entry.  So when replica A tries
                  to acquire a replica B, replica B send a control back to the
                  master(master C) that is updating replica B to abort the session.
                  Master C will continue sending updates for the amount of time
                  specified in the the "release timeout", then it will "yield" its
                  current session so other replicas can acquire that replica.

https://fedorahosted.org/389/ticket/48636

Reviewed by: ?
---
 ldap/schema/01core389.ldif                         |   3 +-
 ldap/servers/plugins/replication/repl5.h           |  14 ++-
 .../plugins/replication/repl5_inc_protocol.c       | 102 ++++++++++++----
 ldap/servers/plugins/replication/repl5_plugins.c   |   3 +-
 ldap/servers/plugins/replication/repl5_replica.c   | 135 +++++++++++++++++----
 .../plugins/replication/repl5_replica_config.c     |  22 ++++
 ldap/servers/plugins/replication/repl_globals.c    |   1 +
 7 files changed, 229 insertions(+), 51 deletions(-)

diff --git a/ldap/schema/01core389.ldif b/ldap/schema/01core389.ldif
index 35d7c4c..ab07a0b 100644
--- a/ldap/schema/01core389.ldif
+++ b/ldap/schema/01core389.ldif
@@ -297,6 +297,7 @@ attributeTypes: ( 2.16.840.1.113730.3.1.2328 NAME 'nsslapd-auditfaillog-list' DE
 attributeTypes: ( 2.16.840.1.113730.3.1.2330 NAME 'nsslapd-logging-backend' DESC 'Netscape defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 X-ORIGIN 'Netscape Directory Server' )
 attributeTypes: ( 2.16.840.1.113730.3.1.2331 NAME 'nsslapd-logging-hr-timestamps-enabled' DESC 'Netscape defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 X-ORIGIN 'Netscape Directory Server' )
 attributeTypes: ( 2.16.840.1.113730.3.1.2332 NAME 'allowWeakDHParam' DESC 'Netscape defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 X-ORIGIN 'Netscape Directory Server' )
+attributeTypes: ( 2.16.840.1.113730.3.1.2333 NAME 'nsds5ReplicaReleaseTimeout' DESC 'Netscape defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.27 SINGLE-VALUE X-ORIGIN 'Netscape Directory Server' )
 #
 # objectclasses
 #
@@ -306,7 +307,7 @@ objectClasses: ( 2.16.840.1.113730.3.2.44 NAME 'nsIndex' DESC 'Netscape defined
 objectClasses: ( 2.16.840.1.113730.3.2.109 NAME 'nsBackendInstance' DESC 'Netscape defined objectclass' SUP top  MUST ( CN ) X-ORIGIN 'Netscape Directory Server' )
 objectClasses: ( 2.16.840.1.113730.3.2.110 NAME 'nsMappingTree' DESC 'Netscape defined objectclass' SUP top  MUST ( CN ) X-ORIGIN 'Netscape Directory Server' )
 objectClasses: ( 2.16.840.1.113730.3.2.104 NAME 'nsContainer' DESC 'Netscape defined objectclass' SUP top  MUST ( CN ) X-ORIGIN 'Netscape Directory Server' )
-objectClasses: ( 2.16.840.1.113730.3.2.108 NAME 'nsDS5Replica' DESC 'Netscape defined objectclass' SUP top  MUST ( nsDS5ReplicaRoot $  nsDS5ReplicaId ) MAY (cn $ nsds5ReplicaPreciseTombstonePurging $ nsds5ReplicaCleanRUV $ nsds5ReplicaAbortCleanRUV $ nsDS5ReplicaType $ nsDS5ReplicaBindDN $ nsState $ nsDS5ReplicaName $ nsDS5Flags $ nsDS5Task $ nsDS5ReplicaReferral $ nsDS5ReplicaAutoReferral $ nsds5ReplicaPurgeDelay $ nsds5ReplicaTombstonePurgeInterval $ nsds5ReplicaChangeCount $ nsds5ReplicaLegacyConsumer $ nsds5ReplicaProtocolTimeout $ nsds5ReplicaBackoffMin $ nsds5ReplicaBackoffMax ) X-ORIGIN 'Netscape Directory Server' )
+objectClasses: ( 2.16.840.1.113730.3.2.108 NAME 'nsDS5Replica' DESC 'Netscape defined objectclass' SUP top  MUST ( nsDS5ReplicaRoot $  nsDS5ReplicaId ) MAY (cn $ nsds5ReplicaPreciseTombstonePurging $ nsds5ReplicaCleanRUV $ nsds5ReplicaAbortCleanRUV $ nsDS5ReplicaType $ nsDS5ReplicaBindDN $ nsState $ nsDS5ReplicaName $ nsDS5Flags $ nsDS5Task $ nsDS5ReplicaReferral $ nsDS5ReplicaAutoReferral $ nsds5ReplicaPurgeDelay $ nsds5ReplicaTombstonePurgeInterval $ nsds5ReplicaChangeCount $ nsds5ReplicaLegacyConsumer $ nsds5ReplicaProtocolTimeout $ nsds5ReplicaBackoffMin $ nsds5ReplicaBackoffMax $ nsds5ReplicaReleaseTimeout ) X-ORIGIN 'Netscape Directory Server' )
 objectClasses: ( 2.16.840.1.113730.3.2.113 NAME 'nsTombstone' DESC 'Netscape defined objectclass' SUP top MAY ( nstombstonecsn $ nsParentUniqueId $ nscpEntryDN ) X-ORIGIN 'Netscape Directory Server' )
 objectClasses: ( 2.16.840.1.113730.3.2.103 NAME 'nsDS5ReplicationAgreement' DESC 'Netscape defined objectclass' SUP top MUST ( cn ) MAY ( nsds5ReplicaCleanRUVNotified $ nsDS5ReplicaHost $ nsDS5ReplicaPort $ nsDS5ReplicaTransportInfo $ nsDS5ReplicaBindDN $ nsDS5ReplicaCredentials $ nsDS5ReplicaBindMethod $ nsDS5ReplicaRoot $ nsDS5ReplicatedAttributeList $ nsDS5ReplicatedAttributeListTotal $ nsDS5ReplicaUpdateSchedule $ nsds5BeginReplicaRefresh $ description $ nsds50ruv $ nsruvReplicaLastModified $ nsds5ReplicaTimeout $ nsds5replicaChangesSentSinceStartup $ nsds5replicaLastUpdateEnd $ nsds5replicaLastUpdateStart $ nsds5replicaLastUpdateStatus $ nsds5replicaUpdateInProgress $ nsds5replicaLastInitEnd $ nsds5ReplicaEnabled $ nsds5replicaLastInitStart $ nsds5replicaLastInitStatus $ nsds5debugreplicatimeout $ nsds5replicaBusyWaitTime $ nsds5ReplicaStripAttrs $ nsds5replicaSessionPauseTime $ nsds5ReplicaProtocolTimeout $ nsds5ReplicaFlowControlWindow $ nsds5ReplicaFlowControlPause $ nsDS5ReplicaWaitForAsyncResults ) X-ORIGIN 'Netscape Directory Server' )
 objectClasses: ( 2.16.840.1.113730.3.2.39 NAME 'nsslapdConfig' DESC 'Netscape defined objectclass' SUP top MAY ( cn ) X-ORIGIN 'Netscape Directory Server' )
diff --git a/ldap/servers/plugins/replication/repl5.h b/ldap/servers/plugins/replication/repl5.h
index 307da82..6f6c81a 100644
--- a/ldap/servers/plugins/replication/repl5.h
+++ b/ldap/servers/plugins/replication/repl5.h
@@ -69,6 +69,10 @@
 #define REPL_ABORT_CLEANRUV_OID "2.16.840.1.113730.3.6.6"
 #define REPL_CLEANRUV_GET_MAXCSN_OID "2.16.840.1.113730.3.6.7"
 #define REPL_CLEANRUV_CHECK_STATUS_OID "2.16.840.1.113730.3.6.8"
+#define REPL_ABORT_SESSION_OID "2.16.840.1.113730.3.6.9"
+#define SESSION_ACQUIRED 0
+#define ABORT_SESSION 1
+#define SESSION_ABORTED 2
 
 #define CLEANRUV_ACCEPTED "accepted"
 #define CLEANRUV_REJECTED "rejected"
@@ -141,6 +145,7 @@ extern const char *type_nsds5ReplicaStripAttrs;
 extern const char *type_nsds5ReplicaFlowControlWindow;
 extern const char *type_nsds5ReplicaFlowControlPause;
 extern const char *type_replicaProtocolTimeout;
+extern const char *type_replicaReleaseTimeout;
 extern const char *type_replicaBackoffMin;
 extern const char *type_replicaBackoffMax;
 extern const char *type_replicaPrecisePurge;
@@ -526,9 +531,9 @@ Replica *replica_new_from_entry (Slapi_Entry *e, char *errortext, PRBool is_add_
 void replica_destroy(void **arg);
 int replica_subentry_update(Slapi_DN *repl_root, ReplicaId rid);
 int replica_subentry_check(Slapi_DN *repl_root, ReplicaId rid);
-PRBool replica_get_exclusive_access(Replica *r, PRBool *isInc, PRUint64 connid, int opid,
-									const char *locking_purl,
-									char **current_purl);
+PRBool replica_get_exclusive_access(Replica *r, PRBool *isInc, PRUint64 connid,
+                                    int opid, const char *locking_purl,
+                                    char **current_purl);
 void replica_relinquish_exclusive_access(Replica *r, PRUint64 connid, int opid);
 PRBool replica_get_tombstone_reap_active(const Replica *r);
 const Slapi_DN *replica_get_root(const Replica *r);
@@ -598,6 +603,8 @@ void replica_update_state (time_t when, void *arg);
 void replica_reset_csn_pl(Replica *r);
 PRUint64 replica_get_protocol_timeout(Replica *r);
 void replica_set_protocol_timeout(Replica *r, PRUint64 timeout);
+PRUint64 replica_get_release_timeout(Replica *r);
+void replica_set_release_timeout(Replica *r, PRUint64 timeout);
 void replica_set_groupdn_checkinterval(Replica *r, int timeout);
 PRUint64 replica_get_backoff_min(Replica *r);
 PRUint64 replica_get_backoff_max(Replica *r);
@@ -609,6 +616,7 @@ void replica_decr_agmt_count(Replica *r);
 PRUint64 replica_get_precise_purging(Replica *r);
 void replica_set_precise_purging(Replica *r, PRUint64 on_off);
 PRBool ignore_error_and_keep_going(int error);
+void replica_check_release_timeout(Replica *r, Slapi_PBlock *pb);
 
 /* The functions below handles the state flag */
 /* Current internal state flags */
diff --git a/ldap/servers/plugins/replication/repl5_inc_protocol.c b/ldap/servers/plugins/replication/repl5_inc_protocol.c
index 927f835..d6fb898 100644
--- a/ldap/servers/plugins/replication/repl5_inc_protocol.c
+++ b/ldap/servers/plugins/replication/repl5_inc_protocol.c
@@ -36,6 +36,11 @@ Perhaps these events should be properties of the main protocol.
 #include "repl5_prot_private.h"
 #include "cl5_api.h"
 
+#include "repl5.h"
+#include "repl5_prot_private.h"
+#include "cl5_api.h"
+#include "slapi-plugin.h"
+
 extern int slapi_log_urp;
 
 /*** from proto-slap.h ***/
@@ -82,6 +87,7 @@ typedef struct result_data
 	int flowcontrol_detection;
 	int result; /* The UPDATE_TRANSIENT_ERROR etc */
 	int WaitForAsyncResults;
+	time_t abort_time;
 } result_data;
 
 /* Various states the incremental protocol can pass through */
@@ -121,6 +127,7 @@ typedef struct result_data
 #define EXAMINE_RUV_PARAM_ERROR 405
 
 #define MAX_CHANGES_PER_SESSION	10000
+
 /*
  * Maximum time to wait between replication sessions. If we
  * don't see any updates for a period equal to this interval,
@@ -240,19 +247,21 @@ repl5_inc_result_threadmain(void *param)
 	Repl_Connection *conn = rd->prp->conn;
 	int finished = 0;
 	int message_id = 0;
+	int yield_session = 0;
 
 	slapi_log_error(SLAPI_LOG_REPL, NULL, "repl5_inc_result_threadmain starting\n");
 	while (!finished) 
 	{
+		LDAPControl **returned_controls = NULL;
 		repl5_inc_operation *op = NULL;
-		int connection_error = 0;
+		ReplicaId replica_id = 0;
 		char *csn_str = NULL; 
 		char *uniqueid = NULL;
-		ReplicaId replica_id = 0;
-		int operation_code = 0;
 		char *ldap_error_string = NULL;
 		time_t time_now = 0;
 		time_t start_time = time( NULL );
+		int connection_error = 0;
+		int operation_code = 0;
 		int backoff_time = 1;
 
 		/* Read the next result */
@@ -264,7 +273,7 @@ repl5_inc_result_threadmain(void *param)
 
 		while (!finished)
 		{
-			conres = conn_read_result_ex(conn, NULL, NULL, NULL, LDAP_RES_ANY, &message_id, 0);
+			conres = conn_read_result_ex(conn, NULL, NULL, &returned_controls, LDAP_RES_ANY, &message_id, 0);
 			slapi_log_error(SLAPI_LOG_REPL, NULL, "repl5_inc_result_threadmain: read result for message_id %d\n", message_id);
 			/* Timeout here means that we didn't block, not a real timeout */
 			if (CONN_TIMEOUT == conres)
@@ -292,9 +301,19 @@ repl5_inc_result_threadmain(void *param)
 					finished = 1;
 				}
 				PR_Unlock(rd->lock);
-			} else
-			{
-				/* Something other than a timeout, so we exit the loop */
+			} else {
+				/*
+				 * Something other than a timeout, so we exit the loop.
+				 * First check if we were told to abort the session
+				 */;
+				Replica *r = (Replica*)object_get_data(rd->prp->replica_object);
+				if (replica_get_release_timeout(r) &&
+				    slapi_control_present(returned_controls,
+				                          REPL_ABORT_SESSION_OID,
+				                          NULL, NULL))
+				{
+					yield_session = 1;
+				}
 				break;
 			}
 		}
@@ -318,21 +337,29 @@ repl5_inc_result_threadmain(void *param)
 			}
 
 			conn_get_error_ex(conn, &operation_code, &connection_error, &ldap_error_string);
-			slapi_log_error(SLAPI_LOG_REPL, NULL, "repl5_inc_result_threadmain: result %d, %d, %d, %d, %s\n", operation_code,connection_error,conres,message_id,ldap_error_string);
-			return_value = repl5_inc_update_from_op_result(rd->prp, conres, connection_error, csn_str, uniqueid, replica_id, &should_finish, &(rd->num_changes_sent));
+			slapi_log_error(SLAPI_LOG_REPL, NULL,
+			                "repl5_inc_result_threadmain: result %d, %d, %d, %d, %s\n",
+			                operation_code,connection_error,conres,message_id,ldap_error_string);
+			return_value = repl5_inc_update_from_op_result(rd->prp, conres, connection_error,
+			                                               csn_str, uniqueid, replica_id, &should_finish,
+			                                               &(rd->num_changes_sent));
 			if (return_value || should_finish)
 			{
-				slapi_log_error(SLAPI_LOG_REPL, NULL, "repl5_inc_result_threadmain: got op result %d should finish %d\n", return_value, should_finish);
+				slapi_log_error(SLAPI_LOG_REPL, NULL,
+				                "repl5_inc_result_threadmain: got op result %d should finish %d\n",
+				                return_value, should_finish);
 				/* If so then we need to take steps to abort the update process */
 				PR_Lock(rd->lock);
 				rd->result = return_value;
-				rd->abort = 1;
+				rd->abort = ABORT_SESSION;
 				PR_Unlock(rd->lock);
-				/* We also need to log the error, including details stored from when the operation was sent */
-				/* we cannot finish yet - we still need to waitfor the pending results, then
-				   the main repl code will shut down this thread */
-				/* we can finish if we have disconnected - in that case, there will be nothing
-				   to read */
+				/*
+				 * We also need to log the error, including details stored from
+				 * when the operation was sent.  We cannot finish yet - we still
+				 * need to wait for the pending results, then the main repl code
+				 * will shut down this thread.  We can finish if we have
+				 * disconnected - in that case, there will be nothing to read
+				 */
 				if (return_value == UPDATE_CONNECTION_LOST) {
 					finished = 1;
 				}
@@ -341,8 +368,16 @@ repl5_inc_result_threadmain(void *param)
 				rd->result = return_value;
 			}
 		}
+
 		/* Should we stop ? */
 		PR_Lock(rd->lock);
+		if (!finished && yield_session && rd->abort != SESSION_ABORTED && rd->abort_time == 0) {
+			rd->abort_time = time( NULL );
+			rd->abort = SESSION_ABORTED;  /* only set the abort time once */
+			slapi_log_error(SLAPI_LOG_REPL, "repl5_inc_result_threadmain",
+			                "Abort control detected, setting abort time...(%s)\n",
+			                agmt_get_long_name(rd->prp->agmt));
+		}
 		if (rd->stop_result_thread) 
 		{
 			finished = 1;
@@ -468,7 +503,8 @@ repl5_inc_waitfor_async_results(result_data *rd)
 		if (rd->last_message_id_received >= rd->last_message_id_sent) {
 			/* If so then we're done */
 			done = 1;
-		} else if (rd->abort && (rd->result == UPDATE_CONNECTION_LOST)) {
+		} else if (rd->abort && (rd->result == UPDATE_CONNECTION_LOST))
+		{
 			done = 1; /* no connection == no more results */
 		}
 		/*
@@ -846,10 +882,10 @@ repl5_inc_run(Private_Repl_Protocol *prp)
                   if (!busywaittime){
                       busywaittime = repl5_get_backoff_min(prp);
                   }
-                  prp_priv->backoff = backoff_new(BACKOFF_FIXED, busywaittime, busywaittime);
+                  prp_priv->backoff = backoff_new(BACKOFF_FIXED, busywaittime , busywaittime);
               } else {
                   prp_priv->backoff = backoff_new(BACKOFF_EXPONENTIAL, repl5_get_backoff_min(prp),
-                		  repl5_get_backoff_max(prp));
+                                                  repl5_get_backoff_max(prp));
               }
               next_state = STATE_BACKOFF;
               backoff_reset(prp_priv->backoff, repl5_inc_backoff_expired, (void *)prp);
@@ -1055,6 +1091,7 @@ repl5_inc_run(Private_Repl_Protocol *prp)
                       } else if (rc == UPDATE_YIELD){
                           dev_debug("repl5_inc_run(STATE_SENDING_UPDATES) -> send_updates = UPDATE_YIELD -> STATE_BACKOFF_START");
                           agmt_set_last_update_status(prp->agmt, 0, 0, "Incremental update succeeded and yielded");
+                          use_busy_backoff_timer = PR_TRUE;
                           next_state = STATE_BACKOFF_START;
                       } else if (rc == UPDATE_TRANSIENT_ERROR){
                           dev_debug("repl5_inc_run(STATE_SENDING_UPDATES) -> send_updates = UPDATE_TRANSIENT_ERROR -> STATE_BACKOFF_START");
@@ -1099,6 +1136,7 @@ repl5_inc_run(Private_Repl_Protocol *prp)
               ruv_destroy(&ruv); ruv = NULL;
           }
           agmt_update_done(prp->agmt, 0);
+
           /* If timed out, close the connection after released the replica */
           release_replica(prp);
           if (rc == UPDATE_TIMEOUT) {
@@ -1681,12 +1719,14 @@ send_updates(Private_Repl_Protocol *prp, RUV *remote_update_vector, PRUint32 *nu
 	}
 	else
 	{
-		int finished = 0;
 		ConnResult replay_crc;
-		char csn_str[CSN_STRSIZE];
+		Replica *replica = (Replica*) object_get_data(prp->replica_object);
 		PRBool subentry_update_needed = PR_FALSE;
+		PRUint64 release_timeout = replica_get_release_timeout(replica);
+		char csn_str[CSN_STRSIZE];
 		int skipped_updates = 0;
 		int fractional_repl;
+		int finished = 0;
 #define FRACTIONAL_SKIPPED_THRESHOLD 100
 
 		/* Start the results reading thread */
@@ -1906,7 +1946,20 @@ send_updates(Private_Repl_Protocol *prp, RUV *remote_update_vector, PRUint32 *nu
 			}
 			PR_Lock(rd->lock);
 			/* See if the result thread has hit a problem */
-			if (!finished && rd->abort)
+
+			if(!finished && rd->abort_time){
+				time_t current_time = time ( NULL );
+				if ((current_time - rd->abort_time) >= release_timeout){
+					rd->result = UPDATE_YIELD;
+					return_value = UPDATE_YIELD;
+					finished = 1;
+					slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
+					                "Aborting send_updates...(%s)\n",
+					                agmt_get_long_name(rd->prp->agmt));
+				}
+			}
+
+			if (!finished && rd->abort == ABORT_SESSION)
 			{
 				return_value = rd->result;
 				finished = 1;
@@ -1916,10 +1969,9 @@ send_updates(Private_Repl_Protocol *prp, RUV *remote_update_vector, PRUint32 *nu
 
 		if (fractional_repl && subentry_update_needed)
 		{
-			Replica *replica;
 			ReplicaId rid = -1; /* Used to create the replica keep alive subentry */
 			Slapi_DN *replarea_sdn = NULL;
-			replica = (Replica*) object_get_data(prp->replica_object);
+
 			if (replica)
 			{
 				rid = replica_get_rid(replica);
@@ -1945,7 +1997,7 @@ send_updates(Private_Repl_Protocol *prp, RUV *remote_update_vector, PRUint32 *nu
 				 * If we already have an error, there is no need to check the
 				 * async result thread anymore.
 				 */
-				if (return_value == UPDATE_NO_MORE_UPDATES)
+				if (return_value == UPDATE_NO_MORE_UPDATES || return_value == UPDATE_YIELD)
 				{
 					/*
 					 * We need to double check that an error hasn't popped up from
diff --git a/ldap/servers/plugins/replication/repl5_plugins.c b/ldap/servers/plugins/replication/repl5_plugins.c
index 3ef3f5f..b331c81 100644
--- a/ldap/servers/plugins/replication/repl5_plugins.c
+++ b/ldap/servers/plugins/replication/repl5_plugins.c
@@ -1077,6 +1077,8 @@ write_changelog_and_ruv (Slapi_PBlock *pb)
 	r = (Replica*)object_get_data (repl_obj);
 	PR_ASSERT (r);
 
+	replica_check_release_timeout(r, pb);
+
 	if (replica_is_flag_set (r, REPLICA_LOG_CHANGES) &&
 		(cl5GetState () == CL5_STATE_OPEN))
 	{
@@ -1365,7 +1367,6 @@ process_postop (Slapi_PBlock *pb)
 	return rc;
 }
 
-
 /*
  * Cancel an operation CSN. This removes it from any CSN pending lists.
  * This function is called when a previously-generated CSN will not
diff --git a/ldap/servers/plugins/replication/repl5_replica.c b/ldap/servers/plugins/replication/repl5_replica.c
index 5de6a49..b5d65ef 100644
--- a/ldap/servers/plugins/replication/repl5_replica.c
+++ b/ldap/servers/plugins/replication/repl5_replica.c
@@ -24,8 +24,8 @@
 
 #define RUV_SAVE_INTERVAL (30 * 1000) /* 30 seconds */
 
-#define REPLICA_RDN				 "cn=replica"
-#define CHANGELOG_RDN            "cn=legacy changelog"
+#define REPLICA_RDN "cn=replica"
+#define CHANGELOG_RDN "cn=legacy changelog"
 
 /*
  * A replica is a locally-held copy of a portion of the DIT.
@@ -69,6 +69,8 @@ struct replica {
 	Slapi_Counter *backoff_max;		/* backoff retry maximum */
 	Slapi_Counter *precise_purging;		/* Enable precise tombstone purging */
 	PRUint64 agmt_count;			/* Number of agmts */
+	Slapi_Counter *release_timeout;		/* The amount of time to wait before releasing active replica */
+	PRUint64 abort_session;			/* Abort the current replica session */
 };
 
 
@@ -202,6 +204,7 @@ replica_new_from_entry (Slapi_Entry *e, char *errortext, PRBool is_add_operation
 
 	/* init the slapi_counter/atomic settings */
 	r->protocol_timeout = slapi_counter_new();
+	r->release_timeout = slapi_counter_new();
 	r->backoff_min = slapi_counter_new();
 	r->backoff_max = slapi_counter_new();
 	r->precise_purging = slapi_counter_new();
@@ -410,6 +413,7 @@ replica_destroy(void **arg)
 	}
 
 	slapi_counter_destroy(&r->protocol_timeout);
+	slapi_counter_destroy(&r->release_timeout);
 	slapi_counter_destroy(&r->backoff_min);
 	slapi_counter_destroy(&r->backoff_max);
 	slapi_counter_destroy(&r->precise_purging);
@@ -588,8 +592,7 @@ replica_subentry_update(Slapi_DN *repl_root, ReplicaId rid)
  */
 PRBool
 replica_get_exclusive_access(Replica *r, PRBool *isInc, PRUint64 connid, int opid,
-							 const char *locking_purl,
-							 char **current_purl)
+							 const char *locking_purl, char **current_purl)
 {
 	PRBool rval = PR_TRUE;
 
@@ -612,6 +615,15 @@ replica_get_exclusive_access(Replica *r, PRBool *isInc, PRUint64 connid, int opi
 		{
 			*current_purl = slapi_ch_strdup(r->locking_purl);
 		}
+		if (!(r->repl_state_flags & REPLICA_TOTAL_IN_PROGRESS) &&
+		    replica_get_release_timeout(r))
+		{
+			/*
+			 * We are not doing a total update, so abort the current session
+			 * so other replicas can acquire this server.
+			 */
+			r->abort_session = ABORT_SESSION;
+		}
 	}
 	else
 	{
@@ -620,14 +632,17 @@ replica_get_exclusive_access(Replica *r, PRBool *isInc, PRUint64 connid, int opi
 						connid, opid,
 						slapi_sdn_get_dn(r->repl_root));
 		r->repl_state_flags |= REPLICA_IN_USE;
+		r->abort_session = SESSION_ACQUIRED;
 		if (isInc && *isInc) 
 		{
 			r->repl_state_flags |= REPLICA_INCREMENTAL_IN_PROGRESS;
 		}
 		else
 		{ 
-			/* if connid or opid != 0, it's a total update */
-			/* Both set to 0 means we're disabling replication */
+			/*
+			 * If connid or opid != 0, it's a total update.
+			 * Both set to 0 means we're disabling replication
+			 */
 			if (connid || opid)
 			{
 				r->repl_state_flags |= REPLICA_TOTAL_IN_PROGRESS;
@@ -655,13 +670,13 @@ replica_relinquish_exclusive_access(Replica *r, PRUint64 connid, int opid)
 	/* check to see if the replica is in use and log a warning if not */
 	if (!(r->repl_state_flags & REPLICA_IN_USE))
 	{
-        slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, 
+		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
 					"conn=%" NSPRIu64 " op=%d repl=\"%s\": "
 					"Replica not in use\n",
 					connid, opid,
 					slapi_sdn_get_dn(r->repl_root));
 	} else {
-		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, 
+		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
 					"conn=%" NSPRIu64 " op=%d repl=\"%s\": "
 					"Released replica held by locking_purl=%s\n",
 					connid, opid,
@@ -973,6 +988,24 @@ replica_get_protocol_timeout(Replica *r)
 	}
 }
 
+PRUint64
+replica_get_release_timeout(Replica *r)
+{
+	if(r){
+		return slapi_counter_get_value(r->release_timeout);
+	} else {
+		return 0;
+	}
+}
+
+void
+replica_set_release_timeout(Replica *r, PRUint64 limit)
+{
+	if(r){
+		slapi_counter_set_value(r->release_timeout, limit);
+	}
+}
+
 void
 replica_set_protocol_timeout(Replica *r, PRUint64 timeout)
 {
@@ -980,6 +1013,7 @@ replica_set_protocol_timeout(Replica *r, PRUint64 timeout)
 		slapi_counter_set_value(r->protocol_timeout, timeout);
 	}
 }
+
 void
 replica_set_groupdn_checkinterval(Replica *r, int interval)
 {
@@ -1067,11 +1101,7 @@ replica_get_legacy_purl (const Replica *r)
     char *purl;
 
     replica_lock(r->repl_lock);
-
-    PR_ASSERT (r->legacy_consumer);
-
     purl = slapi_ch_strdup(r->legacy_purl);
-
     replica_unlock(r->repl_lock);
 
     return purl;
@@ -2001,6 +2031,7 @@ _replica_init_from_config (Replica *r, Slapi_Entry *e, char *errortext)
     int backoff_min;
     int backoff_max;
     int ptimeout = 0;
+    int release_timeout = 0;
     int rc;
 
     PR_ASSERT (r && e);
@@ -2085,6 +2116,14 @@ _replica_init_from_config (Replica *r, Slapi_Entry *e, char *errortext)
         slapi_counter_set_value(r->protocol_timeout, ptimeout);
     }
 
+    /* Get the release timeout */
+    release_timeout = slapi_entry_attr_get_int(e, type_replicaReleaseTimeout);
+    if(release_timeout <= 0){
+        slapi_counter_set_value(r->release_timeout, 0);
+    } else {
+        slapi_counter_set_value(r->release_timeout, release_timeout);
+    }
+
     /* check for precise tombstone purging */
     precise_purging = slapi_entry_attr_get_charptr(e, type_replicaPrecisePurge);
     if(precise_purging){
@@ -4106,21 +4145,21 @@ replica_disable_replication (Replica *r, Object *r_obj)
 	ruv_get_first_id_and_purl(repl_ruv, &junkrid, &p_locking_purl);
 	locking_purl = slapi_ch_strdup(p_locking_purl);
 	p_locking_purl = NULL;
-	repl_ruv = NULL;	
-    while (!replica_get_exclusive_access(r, &isInc, 0, 0, "replica_disable_replication",
+	repl_ruv = NULL;
+	while (!replica_get_exclusive_access(r, &isInc, 0, 0, "replica_disable_replication",
 										 &current_purl)) {
-        if (!isInc) /* already locked, but not by inc update - break */
-            break;
-        isInc = PR_FALSE;
-        slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
+		if (!isInc) /* already locked, but not by inc update - break */
+			break;
+		isInc = PR_FALSE;
+		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
 						"replica_disable_replication: "
 						"replica %s is already locked by (%s) for incoming "
 						"incremental update; sleeping 100ms\n",
-                        slapi_sdn_get_ndn (replica_get_root (r)),
+						slapi_sdn_get_ndn (replica_get_root (r)),
 						current_purl ? current_purl : "unknown");
 		slapi_ch_free_string(&current_purl);
-        DS_Sleep(PR_MillisecondsToInterval(100));
-    }
+		DS_Sleep(PR_MillisecondsToInterval(100));
+	}
 
 	slapi_ch_free_string(&current_purl);
 	slapi_ch_free_string(&locking_purl);
@@ -4358,3 +4397,57 @@ replica_decr_agmt_count(Replica *r)
 		}
 	}
 }
+
+/*
+ * Add the "Abort Replication Session" control to the pblock
+ */
+static void
+replica_add_session_abort_control(Slapi_PBlock *pb)
+{
+	LDAPControl ctrl = {0};
+	BerElement *ber;
+	struct berval *bvp;
+	int rc;
+
+	/* Build the BER payload */
+	if ( (ber = der_alloc()) == NULL ) {
+		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
+			"add_session_abort_control: Failed to create ber\n");
+		return;
+	}
+	rc = ber_printf( ber, "{}");
+	if (rc != -1) {
+		rc = ber_flatten( ber, &bvp );
+	}
+	ber_free( ber, 1 );
+	if ( rc == -1 ) {
+		slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
+			"add_session_abort_control: Failed to flatten ber\n");
+		return;
+	}
+
+	ctrl.ldctl_oid = slapi_ch_strdup( REPL_ABORT_SESSION_OID );
+	ctrl.ldctl_value = *bvp;
+	bvp->bv_val = NULL;
+	ber_bvfree( bvp );
+	slapi_pblock_set(pb, SLAPI_ADD_RESCONTROL, &ctrl);
+
+	slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
+		"add_session_abort_control: abort control successfully added to result\n");
+}
+
+/*
+ * Check if we have exceeded the failed replica acquire limit,
+ * if so, end the replication session.
+ */
+void
+replica_check_release_timeout(Replica *r, Slapi_PBlock *pb)
+{
+	replica_lock(r->repl_lock);
+	if(r->abort_session  == ABORT_SESSION){
+		/* Need to abort this session (just send the control once) */
+		replica_add_session_abort_control(pb);
+		r->abort_session = SESSION_ABORTED;
+	}
+	replica_unlock(r->repl_lock);
+}
diff --git a/ldap/servers/plugins/replication/repl5_replica_config.c b/ldap/servers/plugins/replication/repl5_replica_config.c
index 14020b7..3f6bc9b 100644
--- a/ldap/servers/plugins/replication/repl5_replica_config.c
+++ b/ldap/servers/plugins/replication/repl5_replica_config.c
@@ -406,6 +406,11 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore, Slapi_Entry*
                     if (apply_mods)
                         replica_set_precise_purging(r, 0);
                     }
+                else if (strcasecmp (config_attr, type_replicaReleaseTimeout) == 0 )
+                {
+                    if (apply_mods)
+                        replica_set_release_timeout(r, 0);
+                    }
                 else
                 {
                     *returncode = LDAP_UNWILLING_TO_PERFORM;
@@ -592,6 +597,23 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore, Slapi_Entry*
                         }
                     }
                 }
+                else if (strcasecmp (config_attr, type_replicaReleaseTimeout) == 0 )
+                {
+                    if (apply_mods)
+                    {
+                        PRUint64 val = atoll(config_attr_value);
+
+                        if(val < 0){
+                            *returncode = LDAP_UNWILLING_TO_PERFORM;
+                            PR_snprintf (errortext, SLAPI_DSE_RETURNTEXT_SIZE,
+                                    "attribute %s value (%s) is invalid, must be a number zero or greater.\n",
+                                    config_attr, config_attr_value);
+                            slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "replica_config_modify: %s\n", errortext);
+                            break;
+                        }
+                        replica_set_release_timeout(r, val);
+                    }
+                }
                 else
                 {
                     *returncode = LDAP_UNWILLING_TO_PERFORM;
diff --git a/ldap/servers/plugins/replication/repl_globals.c b/ldap/servers/plugins/replication/repl_globals.c
index 331f839..8b891fb 100644
--- a/ldap/servers/plugins/replication/repl_globals.c
+++ b/ldap/servers/plugins/replication/repl_globals.c
@@ -87,6 +87,7 @@ const char *type_ruvElementUpdatetime = "nsruvReplicaLastModified";
 const char *type_replicaCleanRUV = "nsds5ReplicaCleanRUV";
 const char *type_replicaAbortCleanRUV = "nsds5ReplicaAbortCleanRUV";
 const char *type_replicaProtocolTimeout = "nsds5ReplicaProtocolTimeout";
+const char *type_replicaReleaseTimeout = "nsds5ReplicaReleaseTimeout";
 const char *type_replicaBackoffMin = "nsds5ReplicaBackoffMin";
 const char *type_replicaBackoffMax = "nsds5ReplicaBackoffMax";
 const char *type_replicaPrecisePurge = "nsds5ReplicaPreciseTombstonePurging";
-- 
2.4.11

