s4-drepl: use the partition UDV and hwm for extended getncchanges ops

we find the NC root then load the uptodateness vector and highwater
mark, if available, from there

3 files changed, 75 insertions, 28 deletions

diff --git a/source4/dsdb/repl/drepl_extended.c b/source4/dsdb/repl/drepl_extended.c
index de56cb5ac5a..2c9d1f04e66 100644
--- a/source4/dsdb/repl/drepl_extended.c
+++ b/source4/dsdb/repl/drepl_extended.c
@@ -48,6 +48,8 @@ static WERROR drepl_create_extended_source_dsa(struct dreplsrv_service *service,
 	struct ldb_context *ldb = service->samdb;
 	int ret;
 	WERROR werr;
+	struct ldb_dn *nc_root;
+	struct dreplsrv_partition *p;
 
 	sdsa = talloc_zero(service, struct dreplsrv_partition_source_dsa);
 	W_ERROR_HAVE_NO_MEMORY(sdsa);
@@ -100,7 +102,9 @@ static WERROR drepl_create_extended_source_dsa(struct dreplsrv_service *service,
 		return WERR_NOMEM;
 	}
 
-	sdsa->repsFrom1->highwatermark.highest_usn = min_usn;
+	if (!service->am_rodc) {
+		sdsa->repsFrom1->replica_flags = DRSUAPI_DRS_WRIT_REP;
+	}
 
 	werr = dreplsrv_out_connection_attach(service, sdsa->repsFrom1, &sdsa->conn);
 	if (!W_ERROR_IS_OK(werr)) {
@@ -110,6 +114,39 @@ static WERROR drepl_create_extended_source_dsa(struct dreplsrv_service *service,
 		return werr;
 	}
 
+	ret = dsdb_find_nc_root(service->samdb, sdsa, nc_dn, &nc_root);
+	if (ret != LDB_SUCCESS) {
+		DEBUG(0,(__location__ ": Failed to find nc_root for %s\n",
+			 ldb_dn_get_linearized(nc_dn)));
+		talloc_free(sdsa);
+		return WERR_DS_DRA_INTERNAL_ERROR;
+	}
+
+	/* use the partition uptodateness vector */
+	ret = dsdb_load_udv_v2(service->samdb, nc_root, sdsa->partition,
+			       &sdsa->partition->uptodatevector.cursors,
+			       &sdsa->partition->uptodatevector.count);
+	if (ret != LDB_SUCCESS) {
+		DEBUG(0,(__location__ ": Failed to load UDV for %s\n",
+			 ldb_dn_get_linearized(nc_root)));
+		talloc_free(sdsa);
+		return WERR_DS_DRA_INTERNAL_ERROR;
+	}
+
+	/* find the highwatermark from the partitions list */
+	for (p=service->partitions; p; p=p->next) {
+		if (ldb_dn_compare(p->dn, nc_root) == 0) {
+			struct dreplsrv_partition_source_dsa *s;
+			for (s=p->sources; s; s=s->next) {
+				if (GUID_equal(&s->repsFrom1->source_dsa_obj_guid,
+					       &sdsa->repsFrom1->source_dsa_obj_guid)) {
+					sdsa->repsFrom1->highwatermark = s->repsFrom1->highwatermark;
+				}
+			}
+		}
+	}
+
+
 	*_sdsa = sdsa;
 	return WERR_OK;
 }
diff --git a/source4/dsdb/repl/drepl_out_helpers.c b/source4/dsdb/repl/drepl_out_helpers.c
index 60c4a663450..4f578282ffa 100644
--- a/source4/dsdb/repl/drepl_out_helpers.c
+++ b/source4/dsdb/repl/drepl_out_helpers.c
@@ -297,6 +297,31 @@ static NTSTATUS dreplsrv_get_rodc_partial_attribute_set(struct dreplsrv_service
 	return NT_STATUS_OK;
 }
 
+/*
+  convert from one udv format to the other
+ */
+static WERROR udv_convert(TALLOC_CTX *mem_ctx,
+			  const struct replUpToDateVectorCtr2 *udv,
+			  struct drsuapi_DsReplicaCursorCtrEx *udv_ex)
+{
+	uint32_t i;
+
+	udv_ex->version = 2;
+	udv_ex->reserved1 = 0;
+	udv_ex->reserved2 = 0;
+	udv_ex->count = udv->count;
+	udv_ex->cursors = talloc_array(mem_ctx, struct drsuapi_DsReplicaCursor, udv->count);
+	W_ERROR_HAVE_NO_MEMORY(udv_ex->cursors);
+
+	for (i=0; i<udv->count; i++) {
+		udv_ex->cursors[i].source_dsa_invocation_id = udv->cursors[i].source_dsa_invocation_id;
+		udv_ex->cursors[i].highest_usn = udv->cursors[i].highest_usn;
+	}
+
+	return WERR_OK;
+}
+
+
 static void dreplsrv_op_pull_source_get_changes_trigger(struct tevent_req *req)
 {
 	struct dreplsrv_op_pull_source_state *state = tevent_req_data(req,
@@ -335,6 +360,16 @@ static void dreplsrv_op_pull_source_get_changes_trigger(struct tevent_req *req)
 		return;
 	}
 
+	if (partition->uptodatevector.count != 0 &&
+	    partition->uptodatevector_ex.count == 0) {
+		WERROR werr;
+		werr = udv_convert(partition, &partition->uptodatevector, &partition->uptodatevector_ex);
+		if (!W_ERROR_IS_OK(werr)) {
+			DEBUG(0,(__location__ ": Failed to convert UDV for %s : %s\n",
+				 ldb_dn_get_linearized(partition->dn), nt_errstr(status)));
+		}
+	}
+
 	if (partition->uptodatevector_ex.count == 0) {
 		uptodateness_vector = NULL;
 	} else {
diff --git a/source4/dsdb/repl/drepl_partitions.c b/source4/dsdb/repl/drepl_partitions.c
index caa93e6d544..389a7b4630c 100644
--- a/source4/dsdb/repl/drepl_partitions.c
+++ b/source4/dsdb/repl/drepl_partitions.c
@@ -211,30 +211,6 @@ static WERROR dreplsrv_partition_add_source_dsa(struct dreplsrv_service *s,
 	return WERR_OK;
 }
 
-/*
-  convert from one udv format to the other
- */
-static WERROR udv_convert(TALLOC_CTX *mem_ctx,
-			  const struct replUpToDateVectorCtr2 *udv,
-			  struct drsuapi_DsReplicaCursorCtrEx *udv_ex)
-{
-	uint32_t i;
-
-	udv_ex->version = 2;
-	udv_ex->reserved1 = 0;
-	udv_ex->reserved2 = 0;
-	udv_ex->count = udv->count;
-	udv_ex->cursors = talloc_array(mem_ctx, struct drsuapi_DsReplicaCursor, udv->count);
-	W_ERROR_HAVE_NO_MEMORY(udv_ex->cursors);
-
-	for (i=0; i<udv->count; i++) {
-		udv_ex->cursors[i].source_dsa_invocation_id = udv->cursors[i].source_dsa_invocation_id;
-		udv_ex->cursors[i].highest_usn = udv->cursors[i].highest_usn;
-	}
-
-	return WERR_OK;
-}
-
 WERROR dreplsrv_partition_find_for_nc(struct dreplsrv_service *s,
 				      const struct GUID *nc_guid,
 				      const struct dom_sid *nc_sid,
@@ -352,9 +328,8 @@ static WERROR dreplsrv_refresh_partition(struct dreplsrv_service *s,
 	ZERO_STRUCT(p->uptodatevector_ex);
 
 	ret = dsdb_load_udv_v2(s->samdb, p->dn, p, &p->uptodatevector.cursors, &p->uptodatevector.count);
-	if (ret == LDB_SUCCESS) {
-		status = udv_convert(p, &p->uptodatevector, &p->uptodatevector_ex);
-		W_ERROR_NOT_OK_RETURN(status);
+	if (ret != LDB_SUCCESS) {
+		DEBUG(4,(__location__ ": no UDV available for %s\n", ldb_dn_get_linearized(p->dn)));
 	}
 
 	orf_el = ldb_msg_find_element(r->msgs[0], "repsFrom");