Fix RUV updating code

ldap/servers/plugins/replication/repl5_ruv.c

@@ -199,7 +199,7 @@ ruv_init_from_slapi_attr_and_check_purl(Slapi_Attr *attr, RUV **ruv, ReplicaId *
 						if (NULL != ruve)
 						{
 							/* Is the local purl already in the ruv ? */
-							if ( (*contain_purl==0) && (strncmp(ruve->replica_purl, purl, strlen(purl))==0) )
+							if ( (*contain_purl==0) && ruve->replica_purl && purl && (strncmp(ruve->replica_purl, purl, strlen(purl))==0) )
 							{
 								*contain_purl = ruve->rid;
 							}
@@ -1877,22 +1877,19 @@ ruv_is_newer (Object *sruvobj, Object *cruvobj)
 }
 
 void
-force_csn_update (RUV *ruv, CSN *csn)
+ruv_force_csn_update (RUV *ruv, CSN *csn)
 {
-	CSN *max;
+	CSN *max = NULL;
 	
 	if (ruv != NULL)
 	{
-		
 		ruv_get_max_csn(ruv, &max);
-
 		if (csn_compare(max, csn))
+		{
 			ruv_set_max_csn(ruv, csn, NULL);
-	
+		}
 		csn_free(&max);
 	}	
-
-	
 }
 
 #ifdef TESTING /* Some unit tests for code in this file */

ldap/servers/plugins/replication/repl5_ruv.h

@@ -81,7 +81,7 @@ int ruv_local_contains_supplier(RUV *ruv, ReplicaId rid);
    whether or not an RUV is empty */
 PRBool ruv_has_csns(const RUV *ruv);
 PRBool ruv_is_newer (Object *sruv, Object *cruv);
-void force_csn_update (RUV *ruv, CSN *csn);
+void ruv_force_csn_update (RUV *ruv, CSN *csn);
 #ifdef __cplusplus
 }
 #endif

ldap/servers/plugins/replication/windows_inc_protocol.c

@@ -218,6 +218,11 @@ w_set_pause_and_busy_time(long *pausetime, long *busywaittime)
  *                  schedule_change                        START
  */
 
+/* 
+ * DBDB: what follows is quite possibly the worst code I have ever seen.
+ * Unfortunately we chose not to re-write it when we did the windows sync version.
+ */
+
 /*
  * Main state machine for the incremental protocol. This routine will,
  * under normal circumstances, not return until the protocol is shut
@@ -1088,6 +1093,7 @@ send_updates(Private_Repl_Protocol *prp, RUV *remote_update_vector, PRUint32 *nu
 	int return_value;
 	int rc;
 	CL5ReplayIterator *changelog_iterator = NULL;
+	RUV *current_ruv = ruv_dup(remote_update_vector);
 
 	LDAPDebug( LDAP_DEBUG_TRACE, "=> send_updates\n", 0, 0, 0 );
 
@@ -1293,9 +1299,8 @@ send_updates(Private_Repl_Protocol *prp, RUV *remote_update_vector, PRUint32 *nu
 					/* Positive response received */
 					(*num_changes_sent)++;
 					agmt_inc_last_update_changecount (prp->agmt, csn_get_replicaid(entry.op->csn), 0 /*replayed*/);
-
 					/* bring the consumers (AD) RUV up to date */
-					force_csn_update(remote_update_vector, entry.op->csn);
+					ruv_force_csn_update(current_ruv, entry.op->csn);
 				}
 				break;
 			case CL5_BAD_DATA:
@@ -1352,6 +1357,12 @@ send_updates(Private_Repl_Protocol *prp, RUV *remote_update_vector, PRUint32 *nu
 		w_cl5_operation_parameters_done ( entry.op );
 		cl5DestroyReplayIterator(&changelog_iterator);
 	}
+	/* Save the RUV that we successfully replayed, this ensures that next time we start off at the next changelog record */
+	if (current_ruv)
+	{
+		agmt_set_consumer_ruv(prp->agmt,current_ruv);
+		ruv_destroy(&current_ruv);
+	}
 	LDAPDebug( LDAP_DEBUG_TRACE, "<= send_updates\n", 0, 0, 0 );
 	return return_value;
 }

ldap/servers/plugins/replication/windows_protocol_util.c

@@ -326,6 +326,37 @@ map_dn_values(Private_Repl_Protocol *prp,Slapi_ValueSet *original_values, Slapi_
 	}
 }
 
+static void
+windows_dump_ruvs(Object *supl_ruv_obj, Object *cons_ruv_obj)
+{
+	if (slapi_is_loglevel_set(SLAPI_LOG_REPL))
+	{
+		slapi_log_error(SLAPI_LOG_REPL, NULL, "acquire_replica, supplier RUV:\n");
+		if (supl_ruv_obj) {
+			RUV* sup = NULL;
+			object_acquire(supl_ruv_obj);
+			sup = (RUV*)  object_get_data ( supl_ruv_obj );
+			ruv_dump (sup, "supplier", NULL);
+			object_release(supl_ruv_obj);
+		} else
+		{
+			slapi_log_error(SLAPI_LOG_REPL, NULL, "acquire_replica, supplier RUV = null\n");
+		}
+		slapi_log_error(SLAPI_LOG_REPL, NULL, "acquire_replica, consumer RUV:\n");
+
+		if (cons_ruv_obj) 
+		{
+			RUV* con = NULL;
+			object_acquire(cons_ruv_obj);
+			con =  (RUV*) object_get_data ( cons_ruv_obj );
+			ruv_dump (con,"consumer", NULL);
+			object_release( cons_ruv_obj );
+		} else {
+			slapi_log_error(SLAPI_LOG_REPL, NULL, "acquire_replica, consumer RUV = null\n");
+		}
+	}
+}
+
 /*
  * Acquire exclusive access to a replica. Send a start replication extended
  * operation to the replica. The response will contain a success code, and
@@ -347,8 +378,12 @@ windows_acquire_replica(Private_Repl_Protocol *prp, RUV **ruv, int check_ruv)
 {
   
 	int return_value = ACQUIRE_SUCCESS;
-	ConnResult crc;
-	Repl_Connection *conn;
+	ConnResult crc = 0;
+	Repl_Connection *conn = NULL;
+	Replica *replica = NULL;
+	Object *supl_ruv_obj, *cons_ruv_obj = NULL;
+	PRBool is_newer = PR_FALSE;
+	RUV *r = NULL;
 
 	LDAPDebug( LDAP_DEBUG_TRACE, "=> windows_acquire_replica\n", 0, 0, 0 );
 
@@ -364,79 +399,44 @@ windows_acquire_replica(Private_Repl_Protocol *prp, RUV **ruv, int check_ruv)
         return ACQUIRE_SUCCESS;
     }
 
+	if (NULL != ruv)
 	{
-		Replica *replica;
-		Object *supl_ruv_obj, *cons_ruv_obj;
-		PRBool is_newer = PR_FALSE;
-		RUV *r;
-
-	
-		if (prp->agmt)
-		{		
-			cons_ruv_obj = agmt_get_consumer_ruv(prp->agmt);
-		}
-
-
-
-
-
-		object_acquire(prp->replica_object);
-		replica = object_get_data(prp->replica_object);
-		supl_ruv_obj = replica_get_ruv ( replica );
-
-		/* make a copy of the existing RUV as a starting point 
-		   XXX this is probably a not-so-elegant hack */
-
-		slapi_log_error(SLAPI_LOG_REPL, NULL, "acquire_replica, supplier RUV:\n");
-		if (supl_ruv_obj) {
-		object_acquire(supl_ruv_obj);
-			ruv_dump ((RUV*)  object_get_data ( supl_ruv_obj ), "supplier", NULL);
-			object_release(supl_ruv_obj);
-		}else
-				slapi_log_error(SLAPI_LOG_REPL, NULL, "acquire_replica, supplier RUV = null\n");
-		
-		slapi_log_error(SLAPI_LOG_REPL, NULL, "acquire_replica, consumer RUV:\n");
-
-		if (cons_ruv_obj) 
-		{
-			RUV* con;
-			object_acquire(cons_ruv_obj);
-			con =  (RUV*) object_get_data ( cons_ruv_obj );
-			ruv_dump (con,"consumer", NULL);
-			object_release( cons_ruv_obj );
-		} else {
-			slapi_log_error(SLAPI_LOG_REPL, NULL, "acquire_replica, consumer RUV = null\n");
-		}
+		ruv_destroy ( ruv );
+	}
 
-		is_newer = ruv_is_newer ( supl_ruv_obj, cons_ruv_obj );
-		
-		/* This follows ruv_is_newer, since it's always newer if it's null */
-		if (cons_ruv_obj == NULL) 
-		{
-			RUV *s;
-			s = (RUV*)  object_get_data ( replica_get_ruv ( replica ) );
-			
-			agmt_set_consumer_ruv(prp->agmt, s );
-			object_release ( replica_get_ruv ( replica ) );
-			cons_ruv_obj =  agmt_get_consumer_ruv(prp->agmt);		
-		}
+	object_acquire(prp->replica_object);
+	replica = object_get_data(prp->replica_object);
+	supl_ruv_obj = replica_get_ruv ( replica );
+	cons_ruv_obj = agmt_get_consumer_ruv(prp->agmt);
 
-		r = (RUV*)  object_get_data ( cons_ruv_obj); 
-		*ruv = r;
+	windows_dump_ruvs(supl_ruv_obj,cons_ruv_obj);
+	is_newer = ruv_is_newer ( supl_ruv_obj, cons_ruv_obj );
+	
+	/* Handle the pristine case */
+	if (cons_ruv_obj == NULL) 
+	{
+		/* DBDB: this is all wrong. Need to fix this */
+		RUV *s = NULL;
+		s = (RUV*)  object_get_data ( replica_get_ruv ( replica ) );
 		
-
-
-		if ( supl_ruv_obj ) object_release ( supl_ruv_obj );
-		if ( cons_ruv_obj ) object_release ( cons_ruv_obj );
-		object_release (prp->replica_object);
-		replica = NULL;
-
- 		if (is_newer == PR_FALSE && check_ruv) { 
- 			prp->last_acquire_response_code = NSDS50_REPL_UPTODATE;
-			LDAPDebug( LDAP_DEBUG_TRACE, "<= windows_acquire_replica - ACQUIRE_CONSUMER_WAS_UPTODATE\n", 0, 0, 0 );
- 			return ACQUIRE_CONSUMER_WAS_UPTODATE; 
- 		} 
-	}
+		agmt_set_consumer_ruv(prp->agmt, s );
+		object_release ( replica_get_ruv ( replica ) );
+		cons_ruv_obj = agmt_get_consumer_ruv(prp->agmt);		
+	}
+	r = (RUV*)  object_get_data(cons_ruv_obj); 
+	*ruv = ruv_dup(r);
+
+	if ( supl_ruv_obj ) object_release ( supl_ruv_obj );
+	if ( cons_ruv_obj ) object_release ( cons_ruv_obj );
+	object_release (prp->replica_object);
+	replica = NULL;
+
+	/* Once we get here we have a valid ruv */
+ 	if (is_newer == PR_FALSE && check_ruv) { 
+ 		prp->last_acquire_response_code = NSDS50_REPL_UPTODATE;
+		LDAPDebug( LDAP_DEBUG_TRACE, "<= windows_acquire_replica - ACQUIRE_CONSUMER_WAS_UPTODATE\n", 0, 0, 0 );
+ 		return ACQUIRE_CONSUMER_WAS_UPTODATE; 
+ 	} 
 
 	prp->last_acquire_response_code = NSDS50_REPL_REPLICA_NO_RESPONSE;