Ticket 532 - RUV is not getting updated for both Master and consumer

Bug Description:  If you change the replica type, the RUV does not reflect the change.  This is
                  partly due to the nsState attribute not being updated - specifically the csngen.

Fix Description:  When we change the replica type, update the csn gen, add the new RUV element,
                  delete the old RUV element, clean the changelog RUV, update the state, and notify
                  the agmts.  Then you must reinitialize the agmts after changing the replica type.

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

Reviewed by: richm(Thanks!)

ldap/servers/plugins/replication/repl5.h

@@ -61,6 +61,7 @@
 #include "repl5_ruv.h"
 #include "cl4.h"
 
+#define START_UPDATE_DELAY 2 /* 2 second */
 #define REPLICA_TYPE_WINDOWS 1
 #define REPLICA_TYPE_MULTIMASTER 0
 #define REPL_DIRSYNC_CONTROL_OID "1.2.840.113556.1.4.841"
@@ -558,6 +559,7 @@ int replica_add_by_name (const char *name, Object *replica);
 int replica_delete_by_name (const char *name);
 Object* replica_get_by_name (const char *name);
 void replica_flush(Replica *r);
+void replica_set_csn_assigned(Replica *r);
 void replica_get_referrals(const Replica *r, char ***referrals);
 void replica_set_referrals(Replica *r,const Slapi_ValueSet *vs);
 int replica_update_csngen_state (Replica *r, const RUV *ruv);
@@ -584,6 +586,8 @@ void replica_set_ruv_dirty (Replica *r);
 void replica_write_ruv (Replica *r);
 char *replica_get_dn(Replica *r);
 void replica_check_for_tasks(Replica*r, Slapi_Entry *e);
+void replica_update_state (time_t when, void *arg);
+void replica_reset_csn_pl(Replica *r);
 
 /* The functions below handles the state flag */
 /* Current internal state flags */

ldap/servers/plugins/replication/repl5_plugins.c

@@ -1213,7 +1213,7 @@ write_changelog_and_ruv (Slapi_PBlock *pb)
 			op_params->target_address.uniqueid = slapi_ch_strdup (uniqueid);
 		} 
 
-		if( is_cleaned_rid(csn_get_replicaid(op_params->csn))){
+		if( op_params->csn && is_cleaned_rid(csn_get_replicaid(op_params->csn))){
 			/* this RID has been cleaned */
 			object_release (repl_obj);
 			return 0;

ldap/servers/plugins/replication/repl5_replica.c

@@ -51,7 +51,6 @@
 #include "cl5_api.h"
 
 #define RUV_SAVE_INTERVAL (30 * 1000) /* 30 seconds */
-#define START_UPDATE_DELAY 2 /* 2 second */
 
 #define REPLICA_RDN				 "cn=replica"
 #define CHANGELOG_RDN            "cn=legacy changelog"
@@ -107,7 +106,6 @@ static int _replica_check_validity (const Replica *r);
 static int _replica_init_from_config (Replica *r, Slapi_Entry *e, char *errortext);
 static int _replica_update_entry (Replica *r, Slapi_Entry *e, char *errortext);
 static int _replica_configure_ruv (Replica *r, PRBool isLocked);
-static void _replica_update_state (time_t when, void *arg);
 static char * _replica_get_config_dn (const Slapi_DN *root);
 static char * _replica_type_as_string (const Replica *r);
 /* DBDB, I think this is probably bogus : */
@@ -248,7 +246,7 @@ replica_new_from_entry (Slapi_Entry *e, char *errortext, PRBool is_add_operation
        In that case the updated would fail but nothing bad would happen. The next
        scheduled update would save the state */
 	repl_name = slapi_ch_strdup (r->repl_name);
-	r->repl_eqcxt_rs = slapi_eq_repeat(_replica_update_state, repl_name, 
+	r->repl_eqcxt_rs = slapi_eq_repeat(replica_update_state, repl_name,
                                        current_time () + START_UPDATE_DELAY, RUV_SAVE_INTERVAL);
 
 	if (r->tombstone_reap_interval > 0)
@@ -295,10 +293,17 @@ replica_flush(Replica *r)
 		PR_Unlock(r->repl_lock);
 		/* This function take the Lock Inside */
 		/* And also write the RUV */
-		_replica_update_state((time_t)0, r->repl_name);
+		replica_update_state((time_t)0, r->repl_name);
 	}
 }
 
+void
+replica_set_csn_assigned(Replica *r)
+{
+    PR_Lock(r->repl_lock);
+    r->repl_csn_assigned = PR_TRUE;
+    PR_Unlock(r->repl_lock);
+}
 
 /* 
  * Deallocate a replica. arg should point to the address of a
@@ -928,6 +933,19 @@ replica_set_updatedn (Replica *r, const Slapi_ValueSet *vs, int mod_op)
 	PR_Unlock(r->repl_lock);
 }
 
+void
+replica_reset_csn_pl(Replica *r)
+{
+    PR_Lock(r->repl_lock);
+
+    if (NULL != r->min_csn_pl){
+        csnplFree (&r->min_csn_pl);
+    }
+    r->min_csn_pl = csnplNew();
+
+    PR_Unlock(r->repl_lock);
+}
+
 /* gets current replica generation for this replica */
 char *replica_get_generation (const Replica *r)
 {
@@ -1251,8 +1269,8 @@ replica_set_enabled (Replica *r, PRBool enable)
     {
         if (r->repl_eqcxt_rs == NULL)   /* event is not already registered */
         {
-			repl_name = slapi_ch_strdup (r->repl_name);
-            r->repl_eqcxt_rs = slapi_eq_repeat(_replica_update_state, repl_name, 
+            repl_name = slapi_ch_strdup (r->repl_name);
+            r->repl_eqcxt_rs = slapi_eq_repeat(replica_update_state, repl_name,
                                                current_time() + START_UPDATE_DELAY, RUV_SAVE_INTERVAL);  
         }
     }
@@ -2349,8 +2367,8 @@ done:
 
 /* NOTE - this is the only non-api function that performs locking because
    it is called by the event queue */
-static void 
-_replica_update_state (time_t when, void *arg)
+void
+replica_update_state (time_t when, void *arg)
 {
 	int rc;
 	const char *replica_name = (const char *)arg;
@@ -2420,7 +2438,7 @@ _replica_update_state (time_t when, void *arg)
 	dn = _replica_get_config_dn (r->repl_root);
 	if (NULL == dn) {
 		slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
-			"_replica_update_state: failed to get the config dn for %s\n",
+			"replica_update_state: failed to get the config dn for %s\n",
 			slapi_sdn_get_dn (r->repl_root));
 		PR_Unlock(r->repl_lock);
 		goto done;
@@ -2461,7 +2479,7 @@ _replica_update_state (time_t when, void *arg)
 	slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_RESULT, &rc);
 	if (rc != LDAP_SUCCESS) 
 	{
-		slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "_replica_update_state: "
+		slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "replica_update_state: "
 			"failed to update state of csn generator for replica %s: LDAP "
 			"error - %d\n", slapi_sdn_get_dn(r->repl_root), rc);
 	}

ldap/servers/plugins/replication/repl5_replica_config.c

@@ -500,8 +500,12 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore, Slapi_Entry*
 		if (new_repl_id || new_repl_type)
 		{
 			*returncode = replica_config_change_type_and_id(r, new_repl_type, new_repl_id, errortext, apply_mods);
+			PR_Unlock (s_configLock);
+			replica_update_state(0, (void *)replica_get_name(r));
+			PR_Lock (s_configLock);
 			slapi_ch_free_string(&new_repl_id);
 			slapi_ch_free_string(&new_repl_type);
+			agmtlist_notify_all(pb);
 		}
 	}
 
@@ -808,10 +812,35 @@ replica_config_change_type_and_id (Replica *r, const char *new_type,
 
     if (apply_mods)
     {
+        Object *ruv_obj, *gen_obj;
+        RUV *ruv;
+        CSNGen *gen;
+
+        ruv_obj = replica_get_ruv(r);
+        if(ruv_obj){
+            /* we need to rewrite the repl_csngen with the new rid */
+            ruv = object_get_data (ruv_obj);
+            gen_obj = replica_get_csngen (r);
+            if(gen_obj){
+                const char *purl = multimaster_get_local_purl();
+
+                gen = (CSNGen*) object_get_data (gen_obj);
+                csngen_rewrite_rid(gen, rid);
+                if(purl && type == REPLICA_TYPE_UPDATABLE){
+                    ruv_add_replica(ruv, rid, purl);
+                    replica_reset_csn_pl(r);
+                }
+                ruv_delete_replica(ruv, oldrid);
+                replica_set_ruv_dirty(r);
+                cl5CleanRUV(oldrid);
+                replica_set_csn_assigned(r);
+            }
+            object_release(ruv_obj);
+        }
         replica_set_type (r, type);
-		replica_set_rid(r, rid);
+        replica_set_rid(r, rid);
 
-		/* Set the mapping tree node, and the list of referrals */
+        /* Set the mapping tree node, and the list of referrals */
         /* if this server is a 4.0 consumer the referrals are set by legacy plugin */
         if (!replica_is_legacy_consumer(r))
 		    consumer5_set_mapping_tree_state_for_replica(r, NULL);

ldap/servers/plugins/replication/repl5_ruv.c

@@ -466,7 +466,7 @@ ruv_add_replica (RUV *ruv, ReplicaId rid, const char *replica_purl)
     {
         replica = ruvAddReplicaNoCSN (ruv, rid, replica_purl);
     }
-    
+
     slapi_rwlock_unlock (ruv->lock);
 
     if (replica)
@@ -514,7 +514,7 @@ ruv_add_index_replica (RUV *ruv, ReplicaId rid, const char *replica_purl, int in
     {
         replica = ruvAddIndexReplicaNoCSN (ruv, rid, replica_purl, index);
     }
-    
+
     slapi_rwlock_unlock (ruv->lock);
 
     if (replica)
@@ -1983,14 +1983,14 @@ ruv_move_local_supplier_to_first(RUV *ruv, ReplicaId aRid)
 	
 	PR_ASSERT(ruv);
 
-    slapi_rwlock_wrlock (ruv->lock);
-	
+	slapi_rwlock_wrlock (ruv->lock);
+
 	elem = (RUVElement *)dl_delete(ruv->elements,(const void*)&aRid, ruvReplicaCompare, 0);
 	if (elem) {
 		dl_add_index(ruv->elements, elem, 1);
 		rc = RUV_SUCCESS;
 	}
-	
+
 	slapi_rwlock_unlock (ruv->lock);
 	
 	return rc;

ldap/servers/slapd/csngen.c

@@ -306,11 +306,22 @@ void csngen_abort_csn (CSNGen *gen, const CSN *csn)
     _csngen_call_callbacks (gen, csn, 1);
 }
 
+void csngen_rewrite_rid(CSNGen *gen, ReplicaId rid)
+{
+    if (gen == NULL){
+        return;
+    }
+    slapi_rwlock_wrlock (gen->lock);
+    gen->state.rid = rid;
+    slapi_rwlock_unlock (gen->lock);
+
+}
+
 /* this function should be called when a remote CSN for the same part of
    the dit becomes known to the server (for instance, as part of RUV during
    replication session. In response, the generator would adjust its notion
    of time so that it does not generate smaller csns */
-int csngen_adjust_time (CSNGen *gen, const CSN* csn)
+int csngen_adjust_time(CSNGen *gen, const CSN* csn)
 {
     time_t remote_time, remote_offset, cur_time;
 	PRUint16 remote_seqnum;

ldap/servers/slapd/slapi-private.h

@@ -264,6 +264,8 @@ void csngen_abort_csn (CSNGen *gen, const CSN *csn);
 int csngen_adjust_time (CSNGen *gen, const CSN* csn);
 /* returns PR_TRUE if the csn was generated by this generator and 
    PR_FALSE otherwise. */
+void csngen_rewrite_rid(CSNGen *gen, ReplicaId rid);
+
 PRBool csngen_is_local_csn(const CSNGen *gen, const CSN *csn);
 
 /* returns current state of the generator so that it can be saved in the DIT */