From 49882b927c4b4d74021349d54bd58507a4223de4 Mon Sep 17 00:00:00 2001
From: Ludwig Krispenz <lkrispen@redhat.com>
Date: Mon, 23 May 2016 14:00:13 +0200
Subject: [PATCH] reworked clcach buffer code following design at
 http://www.port389.org/docs/389ds/design/changelog-processing-in-repl-state-sending-updates.html

---
 ldap/servers/plugins/replication/cl5_api.c     | 145 ++----------
 ldap/servers/plugins/replication/cl5_clcache.c | 292 +++++++++++++++----------
 ldap/servers/plugins/replication/cl5_clcache.h |   2 +-
 3 files changed, 201 insertions(+), 238 deletions(-)

diff --git a/ldap/servers/plugins/replication/cl5_api.c b/ldap/servers/plugins/replication/cl5_api.c
index ae23353..3f0ba39 100644
--- a/ldap/servers/plugins/replication/cl5_api.c
+++ b/ldap/servers/plugins/replication/cl5_api.c
@@ -5489,18 +5489,13 @@ static int _cl5PositionCursorForReplay (ReplicaId consumerRID, const RUV *consum
 {
 	CLC_Buffer *clcache = NULL;
 	CL5DBFile *file;
-    int i;
-    CSN **csns = NULL;
     CSN *startCSN = NULL;
-    CSN *minCSN = NULL;
     char csnStr [CSN_STRSIZE];
     int rc = CL5_SUCCESS;
     Object *supplierRuvObj = NULL;
     RUV *supplierRuv = NULL;
-    PRBool newReplica;
     PRBool haveChanges = PR_FALSE;
 	char *agmt_name;
-	ReplicaId rid;
 
     PR_ASSERT (consumerRuv && replica && fileObj && iterator);
 	csnStr[0] = '\0';
@@ -5528,111 +5523,32 @@ static int _cl5PositionCursorForReplay (ReplicaId consumerRID, const RUV *consum
 		ruv_dump (supplierRuv, agmt_name, NULL);
 	}
    
-	/*
-	 * get the sorted list of SupplierMinCSN (if no ConsumerMaxCSN)
-	 * and ConsumerMaxCSN for those RIDs where consumer is not
-	 * up-to-date.
-	 */
-    csns = cl5BuildCSNList (consumerRuv, supplierRuv);
-    if (csns == NULL)
-    {
-        rc = CL5_NOTFOUND;
-        goto done;
-    }
 
-	/* iterate over elements of consumer's (and/or supplier's) ruv */
-    for (i = 0; csns[i]; i++)
-    {
-        CSN *consumerMaxCSN = NULL;
+	/* initialize the changelog buffer and do the initial load */
 
-		rid = csn_get_replicaid(csns[i]);
+	rc = clcache_get_buffer ( &clcache, file->db, consumerRID, consumerRuv, supplierRuv );
+	if ( rc != 0 ) goto done;
 
-		/*
-		 * Skip CSN that is originated from the consumer.
-		 * If RID==65535, the CSN is originated from a
-		 * legacy consumer. In this case the supplier
-		 * and the consumer may have the same RID.
-		 */
-		if ((rid == consumerRID && rid != MAX_REPLICA_ID) || (is_cleaned_rid(rid)) )
-			continue;
-
-        startCSN = csns[i];
-
-		rc = clcache_get_buffer ( &clcache, file->db, consumerRID, consumerRuv, supplierRuv );
-		if ( rc != 0 ) goto done;
-
-	    /* This is the first loading of this iteration. For replicas
-		 * already known to the consumer, we exclude the last entry
-		 * sent to the consumer by using DB_NEXT. However, for
-		 * replicas new to the consumer, we include the first change
-		 * ever generated by that replica.
-		 */
-		newReplica = ruv_get_largest_csn_for_replica (consumerRuv, rid, &consumerMaxCSN);
-		csn_free(&consumerMaxCSN);
-		rc = clcache_load_buffer (clcache, startCSN, (newReplica ? DB_SET : DB_NEXT));
-
-		/* there is a special case which can occur just after migration - in this case,
-		the consumer RUV will contain the last state of the supplier before migration,
-		but the supplier will have an empty changelog, or the supplier changelog will
-		not contain any entries within the consumer min and max CSN - also, since
-		the purge RUV contains no CSNs, the changelog has never been purged
-		ASSUMPTIONS - it is assumed that the supplier had no pending changes to send
-		to any consumers; that is, we can assume that no changes were lost due to
-		either changelog purging or database reload - bug# 603061 - richm@netscape.com
-		*/
-        if ((rc == DB_NOTFOUND) && !ruv_has_csns(file->purgeRUV))
-        {
-            char mincsnStr[CSN_STRSIZE];
-
-            /* use the supplier min csn for the buffer start csn - we know
-               this csn is in our changelog */
-            if ((RUV_SUCCESS == ruv_get_min_csn_ext(supplierRuv, &minCSN, 1 /* ignore cleaned rids */)) &&
-                minCSN)
-            { /* must now free startCSN */
-                if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
-                    csn_as_string(startCSN, PR_FALSE, csnStr);
-                    csn_as_string(minCSN, PR_FALSE, mincsnStr);
-                    slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name_cl, 
-                                    "%s: CSN %s not found and no purging, probably a reinit\n",
-                                    agmt_name, csnStr);
-                    slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name_cl, 
-                                    "%s: Will try to use supplier min CSN %s to load changelog\n",
-                                    agmt_name, mincsnStr);
-                }
-                startCSN = minCSN;
-                rc = clcache_load_buffer (clcache, startCSN, DB_SET);
-            }
-            else
-            {
-                if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
-                    csn_as_string(startCSN, PR_FALSE, csnStr); 
-                    slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name_cl,
-                                    "%s: CSN %s not found and no purging, probably a reinit\n",
-                                    agmt_name, csnStr);
-                    slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name_cl, 
-                                    "%s: Could not get the min csn from the supplier RUV\n",
-                                    agmt_name);
-                }
-                rc = CL5_RUV_ERROR;
-                goto done;
-            }
-        }
+	rc = clcache_load_buffer (clcache, &startCSN);
 
         if (rc == 0) {
-            haveChanges = PR_TRUE;
-            rc = CL5_SUCCESS;
-            if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
-                csn_as_string(startCSN, PR_FALSE, csnStr); 
-                slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name_cl, 
-                                "%s: CSN %s found, position set for replay\n", agmt_name, csnStr);
-            }
-            if (startCSN != csns[i]) {
-                csn_free(&startCSN);
-            }
-            break;
+		haveChanges = PR_TRUE;
+		rc = CL5_SUCCESS;
+		if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
+			csn_as_string(startCSN, PR_FALSE, csnStr);
+			slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name_cl,
+				"%s: CSN %s found, position set for replay\n", agmt_name, csnStr);
+		}
         }
-        else if (rc == DB_NOTFOUND)  /* entry not found */
-        {
+        else if (rc == DB_NOTFOUND)   {
+		/* buffer not loaded.
+		 * either because no changes have to be sent ==> startCSN is NULL
+		 * or the calculated startCSN cannot be found in the changelog
+		 */
+		if (startCSN == NULL) {
+			rc = CL5_NOTFOUND;
+			goto done;
+		}
             /* check whether this csn should be present */
             rc = _cl5CheckMissingCSN (startCSN, supplierRuv, file);
             if (rc == CL5_MISSING_DATA)  /* we should have had the change but we don't */
@@ -5650,17 +5566,6 @@ static int _cl5PositionCursorForReplay (ReplicaId consumerRID, const RUV *consum
                                 "%s: CSN %s not found, we aren't as up to date, or we purged\n", 
                                 agmt_name, csnStr);
             }
-            if (startCSN != csns[i]) {
-                csn_free(&startCSN);
-            }
-            if (rc == CL5_MISSING_DATA)  /* we should have had the change but we don't */
-            {
-                break;
-            }
-            else /* we are not as up to date or we purged */
-            {
-                continue;
-            } 
         }
         else
         {
@@ -5669,15 +5574,10 @@ static int _cl5PositionCursorForReplay (ReplicaId consumerRID, const RUV *consum
 			slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name_cl, 
                             "%s: Failed to retrieve change with CSN %s; db error - %d %s\n", 
                             agmt_name, csnStr, rc, db_strerror(rc));
-            if (startCSN != csns[i]) {
-                csn_free(&startCSN);
-            }
 
             rc = CL5_DB_ERROR;
-            break;
-        }
+    }
 
-    } /* end for */
 
     /* setup the iterator */
     if (haveChanges)
@@ -5709,9 +5609,6 @@ done:
 	if ( clcache )
 		clcache_return_buffer ( &clcache );
 
-    if (csns)
-        cl5DestroyCSNList (&csns);
-
     if (rc != CL5_SUCCESS)
     {
         if (supplierRuvObj)
diff --git a/ldap/servers/plugins/replication/cl5_clcache.c b/ldap/servers/plugins/replication/cl5_clcache.c
index b53d7c0..fb62c0f 100644
--- a/ldap/servers/plugins/replication/cl5_clcache.c
+++ b/ldap/servers/plugins/replication/cl5_clcache.c
@@ -39,6 +39,7 @@
 #define DEFAULT_CLC_BUFFER_COUNT_MAX		0
 #define DEFAULT_CLC_BUFFER_PAGE_COUNT		32
 #define DEFAULT_CLC_BUFFER_PAGE_SIZE		1024
+#define WORK_CLC_BUFFER_PAGE_SIZE 8*DEFAULT_CLC_BUFFER_PAGE_SIZE
 
 enum {
 	CLC_STATE_READY = 0,		/* ready to iterate */
@@ -56,8 +57,9 @@ struct csn_seq_ctrl_block {
 	ReplicaId	rid;				/* RID this block serves */
 	CSN			*consumer_maxcsn;	/* Don't send CSN <= this */
 	CSN			*local_maxcsn;		/* Don't send CSN > this */
-	CSN			*prev_local_maxcsn;	/* */
-	int			state;				/* CLC_STATE_* */
+	CSN			*prev_local_maxcsn;	/* Copy of last state at buffer loading */
+	CSN			*local_mincsn;		/* Used to determin anchor csn*/
+	int			state;			/* CLC_STATE_* */
 };
 
 /*
@@ -70,6 +72,8 @@ struct clc_buffer {
 	ReplicaId	 buf_consumer_rid;	/* help checking threshold csn */
 	const RUV	*buf_consumer_ruv;	/* used to skip change */
 	const RUV	*buf_local_ruv;		/* used to refresh local_maxcsn */
+	int		buf_ignoreConsumerRID;	/* how to handle updates from consumer */
+	int	 	buf_load_cnt;		/* number of loads for session */
 
 	/*
 	 * fields for retriving data from DB
@@ -90,7 +94,6 @@ struct clc_buffer {
 	int			 buf_max_cscbs;
 
 	/* fields for debugging stat */
-	int		 	 buf_load_cnt;		/* number of loads for session */
 	int		 	 buf_record_cnt;	/* number of changes for session */
 	int		 	 buf_record_skipped;	/* number of changes skipped */
 	int		 	 buf_skipped_new_rid;	/* number of changes skipped due to new_rid */
@@ -133,7 +136,8 @@ struct clc_pool {
 static struct clc_pool *_pool = NULL;	/* process's buffer pool */
 
 /* static prototypes */
-static int	clcache_adjust_anchorcsn ( CLC_Buffer *buf );
+static int	clcache_initial_anchorcsn ( CLC_Buffer *buf, int *flag );
+static int	clcache_adjust_anchorcsn ( CLC_Buffer *buf, int *flag );
 static void	clcache_refresh_consumer_maxcsns ( CLC_Buffer *buf );
 static int	clcache_refresh_local_maxcsns ( CLC_Buffer *buf );
 static int	clcache_skip_change ( CLC_Buffer *buf );
@@ -251,8 +255,23 @@ clcache_get_buffer ( CLC_Buffer **buf, DB *db, ReplicaId consumer_rid, const RUV
 	}
 
 	if ( NULL != *buf ) {
+		CSN *c_csn = NULL;
+		CSN *l_csn = NULL;
 		(*buf)->buf_consumer_ruv = consumer_ruv;
 		(*buf)->buf_local_ruv = local_ruv;
+		(*buf)->buf_load_flag = DB_MULTIPLE_KEY;
+		ruv_get_largest_csn_for_replica (consumer_ruv, consumer_rid, &c_csn);
+		ruv_get_largest_csn_for_replica (local_ruv, consumer_rid, &l_csn);
+		if (l_csn && csn_compare(l_csn, c_csn) > 0) {
+			/* the supplier has updates for the consumer RID and
+			 * these updates are newer than on the consumer
+			 */
+			(*buf)->buf_ignoreConsumerRID = 0;
+		} else {
+			(*buf)->buf_ignoreConsumerRID = 1;
+		}
+		csn_free(&c_csn);
+		csn_free(&l_csn);
 	}
 	else {
 		slapi_log_error ( SLAPI_LOG_FATAL, get_thread_private_agmtname(),
@@ -305,36 +324,25 @@ clcache_return_buffer ( CLC_Buffer **buf )
  *		       historic reason.
  */
 int
-clcache_load_buffer ( CLC_Buffer *buf, CSN *anchorcsn, int flag )
+clcache_load_buffer ( CLC_Buffer *buf, CSN **anchorCSN )
 {
 	int rc = 0;
+        int flag = DB_NEXT;
 
+	if (anchorCSN) *anchorCSN = NULL;
 	clcache_refresh_local_maxcsns ( buf );
 
-	/* Set the loading key */
-	if ( anchorcsn ) {
+	if (buf->buf_load_cnt == 0 ) {
 		clcache_refresh_consumer_maxcsns ( buf );
-		buf->buf_load_flag = DB_MULTIPLE_KEY;
-		csn_as_string ( anchorcsn, 0, (char*)buf->buf_key.data );
-		slapi_log_error ( SLAPI_LOG_REPL, buf->buf_agmt_name,
-				"session start: anchorcsn=%s\n", (char*)buf->buf_key.data );
-	}
-	else if ( csn_get_time(buf->buf_current_csn) == 0 ) {
-		/* time == 0 means this csn has never been set */
-		rc = DB_NOTFOUND;
-	}
-	else if ( clcache_adjust_anchorcsn ( buf ) != 0 ) {
-		rc = DB_NOTFOUND;
-	}
-	else {
-		csn_as_string ( buf->buf_current_csn, 0, (char*)buf->buf_key.data );
-		slapi_log_error ( SLAPI_LOG_REPL, buf->buf_agmt_name,
-				"load next: anchorcsn=%s\n", (char*)buf->buf_key.data );
+		rc = clcache_initial_anchorcsn ( buf, &flag );
+        } else {
+		rc = clcache_adjust_anchorcsn ( buf, &flag );
 	}
 
 	if ( rc == 0 ) {
 
 		buf->buf_state = CLC_STATE_READY;
+		if (anchorCSN) *anchorCSN = buf->buf_current_csn;
 		rc = clcache_load_buffer_bulk ( buf, flag );
 
 		/* Reset some flag variables */
@@ -344,21 +352,15 @@ clcache_load_buffer ( CLC_Buffer *buf, CSN *anchorcsn, int flag )
 				buf->buf_cscbs[i]->state = CLC_STATE_READY;
 			}
 		}
-		else if ( anchorcsn ) {
-			/* Report error only when the missing is persistent */
-			if ( buf->buf_missing_csn && csn_compare (buf->buf_missing_csn, anchorcsn) == 0 ) {
-				if (!buf->buf_prev_missing_csn || csn_compare (buf->buf_prev_missing_csn, anchorcsn)) {
-					slapi_log_error ( SLAPI_LOG_FATAL, buf->buf_agmt_name,
-						"Can't locate CSN %s in the changelog (DB rc=%d). If replication stops, the consumer may need to be reinitialized.\n",
-						(char*)buf->buf_key.data, rc );
-					csn_dup_or_init_by_csn (&buf->buf_prev_missing_csn, anchorcsn);
-				}
-			}
-			else {
-				csn_dup_or_init_by_csn (&buf->buf_missing_csn, anchorcsn);
-			}
+		else {
+			slapi_log_error ( SLAPI_LOG_FATAL, buf->buf_agmt_name,
+					"Can't locate CSN %s in the changelog (DB rc=%d). If replication stops, the consumer may need to be reinitialized.\n",
+					(char*)buf->buf_key.data, rc );
 		}
+	} else if (rc == CLC_STATE_DONE) {
+		rc = DB_NOTFOUND;
 	}
+
 	if ( rc != 0 ) {
 		slapi_log_error ( SLAPI_LOG_REPL, buf->buf_agmt_name,
 				"clcache_load_buffer: rc=%d\n", rc );
@@ -483,7 +485,7 @@ clcache_get_next_change ( CLC_Buffer *buf, void **key, size_t *keylen, void **da
 		 * We're done with the current buffer. Now load the next chunk.
 		 */
 		if ( NULL == *key && CLC_STATE_READY == buf->buf_state ) {
-			rc = clcache_load_buffer ( buf, NULL, DB_NEXT );
+			rc = clcache_load_buffer ( buf, NULL );
 			if ( 0 == rc && buf->buf_record_ptr ) {
 				DB_MULTIPLE_KEY_NEXT ( buf->buf_record_ptr, &buf->buf_data,
 								   *key, *keylen, *data, *datalen );
@@ -521,7 +523,6 @@ clcache_refresh_consumer_maxcsns ( CLC_Buffer *buf )
 	int i;
 
 	for ( i = 0; i < buf->buf_num_cscbs; i++ ) {
-		csn_free(&buf->buf_cscbs[i]->consumer_maxcsn);
 		ruv_get_largest_csn_for_replica (
 				buf->buf_consumer_ruv,
 				buf->buf_cscbs[i]->rid,
@@ -538,14 +539,11 @@ clcache_refresh_local_maxcsn ( const ruv_enum_data *rid_data, void *data )
 	int i;
 
 	rid = csn_get_replicaid ( rid_data->csn );
-
-	/*
-	 * No need to create cscb for consumer's RID.
-	 * If RID==65535, the CSN is originated from a
-	 * legacy consumer. In this case the supplier
-	 * and the consumer may have the same RID.
+	/* we do not handle updates originated at the consumer if not required
+	 * and we ignore RID which have been cleaned
 	 */
-	if ( rid == buf->buf_consumer_rid && rid != MAX_REPLICA_ID )
+	if ( (rid == buf->buf_consumer_rid && buf->buf_ignoreConsumerRID) ||
+		is_cleaned_rid(rid) )
 		return rc;
 
 	for ( i = 0; i < buf->buf_num_cscbs; i++ ) {
@@ -564,9 +562,20 @@ clcache_refresh_local_maxcsn ( const ruv_enum_data *rid_data, void *data )
 		}
 		buf->buf_cscbs[i]->rid = rid;
 		buf->buf_num_cscbs++;
+		/* this is the first time we have a local change for the RID
+		 * we need to check what the consumer knows about it.
+		 */
+		ruv_get_largest_csn_for_replica (
+				buf->buf_consumer_ruv,
+				buf->buf_cscbs[i]->rid,
+				&buf->buf_cscbs[i]->consumer_maxcsn );
 	}
 
+	if (buf->buf_cscbs[i]->local_maxcsn)
+	    csn_dup_or_init_by_csn ( &buf->buf_cscbs[i]->prev_local_maxcsn, buf->buf_cscbs[i]->local_maxcsn );
+
 	csn_dup_or_init_by_csn ( &buf->buf_cscbs[i]->local_maxcsn, rid_data->csn );
+	csn_dup_or_init_by_csn ( &buf->buf_cscbs[i]->local_mincsn, rid_data->min_csn );
 
 	if ( buf->buf_cscbs[i]->consumer_maxcsn &&
 		 csn_compare (buf->buf_cscbs[i]->consumer_maxcsn, rid_data->csn) >= 0 ) {
@@ -580,88 +589,147 @@ clcache_refresh_local_maxcsn ( const ruv_enum_data *rid_data, void *data )
 static int
 clcache_refresh_local_maxcsns ( CLC_Buffer *buf )
 {
-	int i;
 
-	for ( i = 0; i < buf->buf_num_cscbs; i++ ) {
-		csn_dup_or_init_by_csn ( &buf->buf_cscbs[i]->prev_local_maxcsn,
-								  buf->buf_cscbs[i]->local_maxcsn );
-	}
 	return ruv_enumerate_elements ( buf->buf_local_ruv, clcache_refresh_local_maxcsn, buf );
 }
 
 /*
  * Algorithm:
  *
- *	1. Snapshot local RUVs;
- *	2. Load buffer;
- *	3. Send to the consumer only those CSNs that are covered
- *	   by the RUVs snapshot taken in the first step;
- *	   All CSNs that are covered by the RUVs snapshot taken in the
- *	   first step are guaranteed in consecutive order for the respected
- *	   RIDs because of the the CSN pending list control;
- *	   A CSN that is not covered by the RUVs snapshot may be out of order
- *	   since it is possible that a smaller CSN might not have committed 
- *	   yet by the time the buffer was loaded.
- *	4. Determine anchorcsn for each RID:
- *
- *	   Case|  Local vs. Buffer | New Local |       Next
- *	       | MaxCSN     MaxCSN |    MaxCSN | Anchor-CSN
- *	   ----+-------------------+-----------+----------------
- *       1 |   Cl    >=   Cb   |     *     | Cb
- *       2 |   Cl    <    Cb   |     Cl    | Cb
- *       3 |   Cl    <    Cb   |     Cl2   | Cl 
- *
- *	5. Determine anchorcsn for next load:
+ *	1. Determine anchorcsn for each RID:
+ *	2. Determine anchorcsn for next load:
  *	   Anchor-CSN = min { all Next-Anchor-CSN, Buffer-MaxCSN }
  */
 static int
-clcache_adjust_anchorcsn ( CLC_Buffer *buf )
+clcache_initial_anchorcsn ( CLC_Buffer *buf, int *flag )
 {
 	PRBool hasChange = PR_FALSE;
 	struct csn_seq_ctrl_block *cscb;
 	int i;
+	CSN *anchorcsn = NULL;
 
 	if ( buf->buf_state == CLC_STATE_READY ) {
 		for ( i = 0; i < buf->buf_num_cscbs; i++ ) {
+			CSN *rid_anchor = NULL;
+			int rid_flag = DB_NEXT;
 			cscb = buf->buf_cscbs[i];
 
-			if ( cscb->state == CLC_STATE_UP_TO_DATE )
-				continue;
+			if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
+					char prevmax[CSN_STRSIZE];
+					char local[CSN_STRSIZE];
+					char curr[CSN_STRSIZE];
+					char conmaxcsn[CSN_STRSIZE];
+					csn_as_string(cscb->prev_local_maxcsn, 0, prevmax);
+					csn_as_string(cscb->local_maxcsn, 0, local);
+					csn_as_string(buf->buf_current_csn, 0, curr);
+					csn_as_string(cscb->consumer_maxcsn, 0, conmaxcsn);
+					slapi_log_error(SLAPI_LOG_REPL, "clcache_initial_anchorcsn" ,
+									"%s - (cscb %d - state %d) - csnPrevMax (%s) "
+									"csnMax (%s) csnBuf (%s) csnConsumerMax (%s)\n",
+									buf->buf_agmt_name, i, cscb->state, prevmax, local,
+									curr, conmaxcsn);
+			}
 
-			/*
-			 * Case 3 unsafe ruv change: next buffer load should start
-			 * from where the maxcsn in the old ruv was. Since each
-			 * cscb has remembered the maxcsn sent to the consumer,
-			 * CSNs that may be loaded again could easily be skipped.
-			 */
-			if ( cscb->prev_local_maxcsn &&
-				 csn_compare (cscb->prev_local_maxcsn, buf->buf_current_csn) < 0 &&
-				 csn_compare (cscb->local_maxcsn, cscb->prev_local_maxcsn) != 0 ) {
+			if (cscb->consumer_maxcsn == NULL) {
+				/* the consumer hasn't seen changes for this RID */
+				rid_anchor = cscb->local_mincsn;
+				rid_flag = DB_SET;
+			} else if ( csn_compare (cscb->local_maxcsn, cscb->consumer_maxcsn) > 0 ) {
+				rid_anchor = cscb->consumer_maxcsn;
+			}
+
+			if (rid_anchor && (anchorcsn == NULL ||
+			    ( csn_compare(rid_anchor, anchorcsn) < 0))) {
+				anchorcsn = rid_anchor;
+				*flag = rid_flag;
 				hasChange = PR_TRUE;
-				cscb->state = CLC_STATE_READY;
-				csn_init_by_csn ( buf->buf_current_csn, cscb->prev_local_maxcsn );
-				csn_as_string ( cscb->prev_local_maxcsn, 0, (char*)buf->buf_key.data );
-				slapi_log_error ( SLAPI_LOG_REPL, buf->buf_agmt_name,
-						"adjust anchor csn upon %s\n",
-						( cscb->state == CLC_STATE_CSN_GT_RUV ? "out of sequence csn" : "unsafe ruv change") );
-				continue;
 			}
 
-			/*
-			 * check if there are still changes to send for this RID
-			 * Assume we had compared the local maxcsn and the consumer
-			 * max csn before this function was called and hence the
-			 * cscb->state had been set accordingly.
-			 */ 
-			if ( hasChange == PR_FALSE &&
-				 csn_compare (cscb->local_maxcsn, buf->buf_current_csn) > 0 ) {
+
+		}
+	}
+
+	if ( !hasChange ) {
+		buf->buf_state = CLC_STATE_DONE;
+	} else {
+		csn_init_by_csn(buf->buf_current_csn, anchorcsn);
+		csn_as_string(buf->buf_current_csn, 0, (char *)buf->buf_key.data);
+		slapi_log_error(SLAPI_LOG_REPL, "clcache_initial_anchorcsn",
+						"anchor is now: %s\n", (char *)buf->buf_key.data);
+	}
+
+	return buf->buf_state;
+}
+
+static int
+clcache_adjust_anchorcsn ( CLC_Buffer *buf, int *flag )
+{
+	PRBool hasChange = PR_FALSE;
+	struct csn_seq_ctrl_block *cscb;
+	int i;
+	CSN *anchorcsn = NULL;
+
+	if ( buf->buf_state == CLC_STATE_READY ) {
+		for ( i = 0; i < buf->buf_num_cscbs; i++ ) {
+			CSN *rid_anchor = NULL;
+			int rid_flag = DB_NEXT;
+			cscb = buf->buf_cscbs[i];
+
+			if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
+					char prevmax[CSN_STRSIZE];
+					char local[CSN_STRSIZE];
+					char curr[CSN_STRSIZE];
+					char conmaxcsn[CSN_STRSIZE];
+					csn_as_string(cscb->prev_local_maxcsn, 0, prevmax);
+					csn_as_string(cscb->local_maxcsn, 0, local);
+					csn_as_string(buf->buf_current_csn, 0, curr);
+					csn_as_string(cscb->consumer_maxcsn, 0, conmaxcsn);
+					slapi_log_error(SLAPI_LOG_REPL, "clcache_adjust_anchorcsn" ,
+									"%s - (cscb %d - state %d) - csnPrevMax (%s) "
+									"csnMax (%s) csnBuf (%s) csnConsumerMax (%s)\n",
+									buf->buf_agmt_name, i, cscb->state, prevmax, local,
+									curr, conmaxcsn);
+			}
+
+			if (csn_compare (cscb->local_maxcsn, cscb->prev_local_maxcsn) == 0 ||
+			    csn_compare (cscb->prev_local_maxcsn, buf->buf_current_csn) > 0 ) {
+				if (csn_compare (cscb->local_maxcsn, cscb->consumer_maxcsn) > 0 ) {
+					rid_anchor = buf->buf_current_csn;
+				}
+			} else {
+				/* prev local max csn < csnBuffer AND different from local maxcsn */
+				if (cscb->prev_local_maxcsn == NULL) {
+					if (cscb->consumer_maxcsn == NULL) {
+						/* the consumer hasn't seen changes for this RID */
+						rid_anchor = cscb->local_mincsn;
+						rid_flag = DB_SET;
+					} else if ( csn_compare (cscb->local_maxcsn, cscb->consumer_maxcsn) > 0 ) {
+						rid_anchor = cscb->consumer_maxcsn;
+					}
+				} else {
+					/* csnPrevMaxSup > 0 */
+					rid_anchor = cscb->consumer_maxcsn;
+				}
+			}
+
+			if (rid_anchor && (anchorcsn == NULL ||
+			    ( csn_compare(rid_anchor, anchorcsn) < 0))) {
+				anchorcsn = rid_anchor;
+				*flag = rid_flag;
 				hasChange = PR_TRUE;
 			}
+
+
 		}
 	}
 
 	if ( !hasChange ) {
 		buf->buf_state = CLC_STATE_DONE;
+	} else {
+		csn_init_by_csn(buf->buf_current_csn, anchorcsn);
+		csn_as_string(buf->buf_current_csn, 0, (char *)buf->buf_key.data);
+		slapi_log_error(SLAPI_LOG_REPL, "clcache_adjust_anchorcsn",
+						"anchor is now: %s\n", (char *)buf->buf_key.data);
 	}
 
 	return buf->buf_state;
@@ -675,7 +743,6 @@ clcache_skip_change ( CLC_Buffer *buf )
 	int skip = 1;
 	int i;
 	char buf_cur_csn_str[CSN_STRSIZE];
-	char oth_csn_str[CSN_STRSIZE];
 
 	do {
 
@@ -688,25 +755,14 @@ clcache_skip_change ( CLC_Buffer *buf )
 		 * legacy consumer. In this case the supplier
 		 * and the consumer may have the same RID.
 		 */
-		if (rid == buf->buf_consumer_rid && rid != MAX_REPLICA_ID){
-			CSN *cons_maxcsn = NULL;
-
-			ruv_get_max_csn(buf->buf_consumer_ruv, &cons_maxcsn);
-			if ( csn_compare ( buf->buf_current_csn, cons_maxcsn) > 0 ) {
-				/*
-				 *  The consumer must have been "restored" and needs this newer update.
-				 */
-				skip = 0;
-			} else if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
+		if (rid == buf->buf_consumer_rid && buf->buf_ignoreConsumerRID){
+			if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
 				csn_as_string(buf->buf_current_csn, 0, buf_cur_csn_str);
-				csn_as_string(cons_maxcsn, 0, oth_csn_str);
 				slapi_log_error(SLAPI_LOG_REPL, buf->buf_agmt_name,
-					"Skipping update because the changelog buffer current csn [%s] is "
-				        "less than or equal to the consumer max csn [%s]\n",
-				        buf_cur_csn_str, oth_csn_str);
+					"Skipping update because the consumer with Rid: [%d] is "
+				        "ignored\n", rid);
 				buf->buf_skipped_csn_gt_cons_maxcsn++;
 			}
-			csn_free(&cons_maxcsn);
 			break;
 		}
 
@@ -821,6 +877,7 @@ clcache_free_cscb ( struct csn_seq_ctrl_block ** cscb )
 	csn_free ( & (*cscb)->consumer_maxcsn );
 	csn_free ( & (*cscb)->local_maxcsn );
 	csn_free ( & (*cscb)->prev_local_maxcsn );
+	csn_free ( & (*cscb)->local_mincsn );
 	slapi_ch_free ( (void **) cscb );
 }
 
@@ -1003,6 +1060,15 @@ clcache_cursor_get ( DBC *cursor, CLC_Buffer *buf, int flag )
 {
 	int rc;
 
+	if (buf->buf_data.ulen > WORK_CLC_BUFFER_PAGE_SIZE) {
+		/*
+		 * The buffer size had to be increased,
+		 * reset it to a smaller working size,
+		 * if not sufficient it will be increased again
+		 */
+		buf->buf_data.ulen = WORK_CLC_BUFFER_PAGE_SIZE;
+	}
+
 	rc = cursor->c_get ( cursor,
 						 & buf->buf_key,
 						 & buf->buf_data,
diff --git a/ldap/servers/plugins/replication/cl5_clcache.h b/ldap/servers/plugins/replication/cl5_clcache.h
index 4c459ab..75b2817 100644
--- a/ldap/servers/plugins/replication/cl5_clcache.h
+++ b/ldap/servers/plugins/replication/cl5_clcache.h
@@ -23,7 +23,7 @@ typedef struct clc_buffer CLC_Buffer;
 int	 clcache_init ( DB_ENV **dbenv );
 void clcache_set_config ();
 int	 clcache_get_buffer ( CLC_Buffer **buf, DB *db, ReplicaId consumer_rid, const RUV *consumer_ruv, const RUV *local_ruv );
-int	 clcache_load_buffer ( CLC_Buffer *buf, CSN *startCSN, int flag );
+int	 clcache_load_buffer ( CLC_Buffer *buf, CSN **anchorCSN );
 void clcache_return_buffer ( CLC_Buffer **buf );
 int	 clcache_get_next_change ( CLC_Buffer *buf, void **key, size_t *keylen, void **data, size_t *datalen, CSN **csn );
 void clcache_destroy ();
-- 
2.4.3

