7 files changed, 598 insertions, 178 deletions

diff --git a/source4/dsdb/repl/drepl_out_helpers.c b/source4/dsdb/repl/drepl_out_helpers.c
index 722db4f8ee..a4f5d1faec 100644
--- a/source4/dsdb/repl/drepl_out_helpers.c
+++ b/source4/dsdb/repl/drepl_out_helpers.c
@@ -202,81 +202,90 @@ NTSTATUS dreplsrv_out_drsuapi_recv(struct tevent_req *req)
 }
 
 struct dreplsrv_op_pull_source_state {
-	struct composite_context *creq;
-
 	struct dreplsrv_out_operation *op;
-
-	struct dreplsrv_drsuapi_connection *drsuapi;
-
-	bool have_all;
-
-	uint32_t ctr_level;
-	struct drsuapi_DsGetNCChangesCtr1 *ctr1;
-	struct drsuapi_DsGetNCChangesCtr6 *ctr6;
 };
 
 static void dreplsrv_op_pull_source_connect_done(struct tevent_req *subreq);
 
-struct composite_context *dreplsrv_op_pull_source_send(struct dreplsrv_out_operation *op)
+struct tevent_req *dreplsrv_op_pull_source_send(TALLOC_CTX *mem_ctx,
+						struct tevent_context *ev,
+						struct dreplsrv_out_operation *op)
 {
-	struct composite_context *c;
-	struct dreplsrv_op_pull_source_state *st;
+	struct tevent_req *req;
+	struct dreplsrv_op_pull_source_state *state;
 	struct tevent_req *subreq;
 
-	c = composite_create(op, op->service->task->event_ctx);
-	if (c == NULL) return NULL;
-
-	st = talloc_zero(c, struct dreplsrv_op_pull_source_state);
-	if (composite_nomem(st, c)) return c;
+	req = tevent_req_create(mem_ctx, &state,
+				struct dreplsrv_op_pull_source_state);
+	if (req == NULL) {
+		return NULL;
+	}
 
-	st->creq	= c;
-	st->op		= op;
+	state->op = op;
 
-	subreq = dreplsrv_out_drsuapi_send(st,
-					   op->service->task->event_ctx,
-					   op->source_dsa->conn);
-	if (composite_nomem(subreq, c)) return c;
-	tevent_req_set_callback(subreq, dreplsrv_op_pull_source_connect_done, st);
+	subreq = dreplsrv_out_drsuapi_send(state, ev, op->source_dsa->conn);
+	if (tevent_req_nomem(subreq, req)) {
+		return tevent_req_post(req, ev);
+	}
+	tevent_req_set_callback(subreq, dreplsrv_op_pull_source_connect_done, req);
 
-	return c;
+	return req;
 }
 
-static void dreplsrv_op_pull_source_get_changes_send(struct dreplsrv_op_pull_source_state *st);
+static void dreplsrv_op_pull_source_get_changes_trigger(struct tevent_req *req);
 
 static void dreplsrv_op_pull_source_connect_done(struct tevent_req *subreq)
 {
-	struct dreplsrv_op_pull_source_state *st = tevent_req_callback_data(subreq,
-						   struct dreplsrv_op_pull_source_state);
-	struct composite_context *c = st->creq;
+	struct tevent_req *req = tevent_req_callback_data(subreq,
+				 struct tevent_req);
+	NTSTATUS status;
 
-	c->status = dreplsrv_out_drsuapi_recv(subreq);
+	status = dreplsrv_out_drsuapi_recv(subreq);
 	TALLOC_FREE(subreq);
-	if (!composite_is_ok(c)) return;
+	if (tevent_req_nterror(req, status)) {
+		return;
+	}
 
-	dreplsrv_op_pull_source_get_changes_send(st);
+	dreplsrv_op_pull_source_get_changes_trigger(req);
 }
 
-static void dreplsrv_op_pull_source_get_changes_recv(struct rpc_request *req);
+static void dreplsrv_op_pull_source_get_changes_done(struct rpc_request *rreq);
 
-static void dreplsrv_op_pull_source_get_changes_send(struct dreplsrv_op_pull_source_state *st)
+static void dreplsrv_op_pull_source_get_changes_trigger(struct tevent_req *req)
 {
-	struct composite_context *c = st->creq;
-	struct repsFromTo1 *rf1 = st->op->source_dsa->repsFrom1;
-	struct dreplsrv_service *service = st->op->service;
-	struct dreplsrv_partition *partition = st->op->source_dsa->partition;
-	struct dreplsrv_drsuapi_connection *drsuapi = st->op->source_dsa->conn->drsuapi;
-	struct rpc_request *req;
+	struct dreplsrv_op_pull_source_state *state = tevent_req_data(req,
+						      struct dreplsrv_op_pull_source_state);
+	struct repsFromTo1 *rf1 = state->op->source_dsa->repsFrom1;
+	struct dreplsrv_service *service = state->op->service;
+	struct dreplsrv_partition *partition = state->op->source_dsa->partition;
+	struct dreplsrv_drsuapi_connection *drsuapi = state->op->source_dsa->conn->drsuapi;
+	struct rpc_request *rreq;
 	struct drsuapi_DsGetNCChanges *r;
+	struct drsuapi_DsReplicaCursorCtrEx *uptodateness_vector;
 
-	r = talloc(st, struct drsuapi_DsGetNCChanges);
-	if (composite_nomem(r, c)) return;
+	r = talloc(state, struct drsuapi_DsGetNCChanges);
+	if (tevent_req_nomem(r, req)) {
+		return;
+	}
 
 	r->out.level_out = talloc(r, int32_t);
-	if (composite_nomem(r->out.level_out, c)) return;
+	if (tevent_req_nomem(r->out.level_out, req)) {
+		return;
+	}
 	r->in.req = talloc(r, union drsuapi_DsGetNCChangesRequest);
-	if (composite_nomem(r->in.req, c)) return;
+	if (tevent_req_nomem(r->in.req, req)) {
+		return;
+	}
 	r->out.ctr = talloc(r, union drsuapi_DsGetNCChangesCtr);
-	if (composite_nomem(r->out.ctr, c)) return;
+	if (tevent_req_nomem(r->out.ctr, req)) {
+		return;
+	}
+
+	if (partition->uptodatevector_ex.count == 0) {
+		uptodateness_vector = NULL;
+	} else {
+		uptodateness_vector = &partition->uptodatevector_ex;
+	}
 
 	r->in.bind_handle	= &drsuapi->bind_handle;
 	if (drsuapi->remote_info28.supported_extensions & DRSUAPI_SUPPORTED_EXTENSION_GETCHGREQ_V8) {
@@ -285,12 +294,12 @@ static void dreplsrv_op_pull_source_get_changes_send(struct dreplsrv_op_pull_sou
 		r->in.req->req8.source_dsa_invocation_id= rf1->source_dsa_invocation_id;
 		r->in.req->req8.naming_context		= &partition->nc;
 		r->in.req->req8.highwatermark		= rf1->highwatermark;
-		r->in.req->req8.uptodateness_vector	= NULL;/*&partition->uptodatevector_ex;*/
+		r->in.req->req8.uptodateness_vector	= uptodateness_vector;
 		r->in.req->req8.replica_flags		= rf1->replica_flags;
 		r->in.req->req8.max_object_count	= 133;
 		r->in.req->req8.max_ndr_size		= 1336811;
-		r->in.req->req8.extended_op		= DRSUAPI_EXOP_NONE;
-		r->in.req->req8.fsmo_info		= 0;
+		r->in.req->req8.extended_op		= state->op->extended_op;
+		r->in.req->req8.fsmo_info		= state->op->fsmo_info;
 		r->in.req->req8.partial_attribute_set	= NULL;
 		r->in.req->req8.partial_attribute_set_ex= NULL;
 		r->in.req->req8.mapping_ctr.num_mappings= 0;
@@ -301,40 +310,50 @@ static void dreplsrv_op_pull_source_get_changes_send(struct dreplsrv_op_pull_sou
 		r->in.req->req5.source_dsa_invocation_id= rf1->source_dsa_invocation_id;
 		r->in.req->req5.naming_context		= &partition->nc;
 		r->in.req->req5.highwatermark		= rf1->highwatermark;
-		r->in.req->req5.uptodateness_vector	= NULL;/*&partition->uptodatevector_ex;*/
+		r->in.req->req5.uptodateness_vector	= uptodateness_vector;
 		r->in.req->req5.replica_flags		= rf1->replica_flags;
 		r->in.req->req5.max_object_count	= 133;
 		r->in.req->req5.max_ndr_size		= 1336770;
-		r->in.req->req5.extended_op		= DRSUAPI_EXOP_NONE;
-		r->in.req->req5.fsmo_info		= 0;
+		r->in.req->req5.extended_op		= state->op->extended_op;
+		r->in.req->req5.fsmo_info		= state->op->fsmo_info;
 	}
 
-	req = dcerpc_drsuapi_DsGetNCChanges_send(drsuapi->pipe, r, r);
-	composite_continue_rpc(c, req, dreplsrv_op_pull_source_get_changes_recv, st);
+#if 0
+	NDR_PRINT_IN_DEBUG(drsuapi_DsGetNCChanges, r);
+#endif
+
+	rreq = dcerpc_drsuapi_DsGetNCChanges_send(drsuapi->pipe, r, r);
+	if (tevent_req_nomem(rreq, req)) {
+		return;
+	}
+	composite_continue_rpc(NULL, rreq, dreplsrv_op_pull_source_get_changes_done, req);
 }
 
-static void dreplsrv_op_pull_source_apply_changes_send(struct dreplsrv_op_pull_source_state *st,
-						       struct drsuapi_DsGetNCChanges *r,
-						       uint32_t ctr_level,
-						       struct drsuapi_DsGetNCChangesCtr1 *ctr1,
-						       struct drsuapi_DsGetNCChangesCtr6 *ctr6);
+static void dreplsrv_op_pull_source_apply_changes_trigger(struct tevent_req *req,
+						          struct drsuapi_DsGetNCChanges *r,
+						          uint32_t ctr_level,
+						          struct drsuapi_DsGetNCChangesCtr1 *ctr1,
+						          struct drsuapi_DsGetNCChangesCtr6 *ctr6);
 
-static void dreplsrv_op_pull_source_get_changes_recv(struct rpc_request *req)
+static void dreplsrv_op_pull_source_get_changes_done(struct rpc_request *rreq)
 {
-	struct dreplsrv_op_pull_source_state *st = talloc_get_type(req->async.private_data,
-						   struct dreplsrv_op_pull_source_state);
-	struct composite_context *c = st->creq;
-	struct drsuapi_DsGetNCChanges *r = talloc_get_type(req->ndr.struct_ptr,
+	struct tevent_req *req = talloc_get_type(rreq->async.private_data,
+						 struct tevent_req);
+	NTSTATUS status;
+	struct drsuapi_DsGetNCChanges *r = talloc_get_type(rreq->ndr.struct_ptr,
 					   struct drsuapi_DsGetNCChanges);
 	uint32_t ctr_level = 0;
 	struct drsuapi_DsGetNCChangesCtr1 *ctr1 = NULL;
 	struct drsuapi_DsGetNCChangesCtr6 *ctr6 = NULL;
 
-	c->status = dcerpc_ndr_request_recv(req);
-	if (!composite_is_ok(c)) return;
+	status = dcerpc_ndr_request_recv(rreq);
+	if (tevent_req_nterror(req, status)) {
+		return;
+	}
 
 	if (!W_ERROR_IS_OK(r->out.result)) {
-		composite_error(c, werror_to_ntstatus(r->out.result));
+		status = werror_to_ntstatus(r->out.result);
+		tevent_req_nterror(req, status);
 		return;
 	}
 
@@ -361,38 +380,42 @@ static void dreplsrv_op_pull_source_get_changes_recv(struct rpc_request *req)
 		ctr_level = 6;
 		ctr6 = &r->out.ctr->ctr7.ctr.xpress6.ts->ctr6;
 	} else {
-		composite_error(c, werror_to_ntstatus(WERR_BAD_NET_RESP));
+		status = werror_to_ntstatus(WERR_BAD_NET_RESP);
+		tevent_req_nterror(req, status);
 		return;
 	}
 
 	if (!ctr1 && !ctr6) {
-		composite_error(c, werror_to_ntstatus(WERR_BAD_NET_RESP));
+		status = werror_to_ntstatus(WERR_BAD_NET_RESP);
+		tevent_req_nterror(req, status);
 		return;
 	}
 
 	if (ctr_level == 6) {
 		if (!W_ERROR_IS_OK(ctr6->drs_error)) {
-			composite_error(c, werror_to_ntstatus(ctr6->drs_error));
+			status = werror_to_ntstatus(ctr6->drs_error);
+			tevent_req_nterror(req, status);
 			return;
 		}
 	}
 
-	dreplsrv_op_pull_source_apply_changes_send(st, r, ctr_level, ctr1, ctr6);
+	dreplsrv_op_pull_source_apply_changes_trigger(req, r, ctr_level, ctr1, ctr6);
 }
 
-static void dreplsrv_update_refs_send(struct dreplsrv_op_pull_source_state *st);
+static void dreplsrv_update_refs_trigger(struct tevent_req *req);
 
-static void dreplsrv_op_pull_source_apply_changes_send(struct dreplsrv_op_pull_source_state *st,
-						       struct drsuapi_DsGetNCChanges *r,
-						       uint32_t ctr_level,
-						       struct drsuapi_DsGetNCChangesCtr1 *ctr1,
-						       struct drsuapi_DsGetNCChangesCtr6 *ctr6)
+static void dreplsrv_op_pull_source_apply_changes_trigger(struct tevent_req *req,
+							  struct drsuapi_DsGetNCChanges *r,
+							  uint32_t ctr_level,
+							  struct drsuapi_DsGetNCChangesCtr1 *ctr1,
+							   struct drsuapi_DsGetNCChangesCtr6 *ctr6)
 {
-	struct composite_context *c = st->creq;
-	struct repsFromTo1 rf1 = *st->op->source_dsa->repsFrom1;
-	struct dreplsrv_service *service = st->op->service;
-	struct dreplsrv_partition *partition = st->op->source_dsa->partition;
-	struct dreplsrv_drsuapi_connection *drsuapi = st->op->source_dsa->conn->drsuapi;
+	struct dreplsrv_op_pull_source_state *state = tevent_req_data(req,
+						      struct dreplsrv_op_pull_source_state);
+	struct repsFromTo1 rf1 = *state->op->source_dsa->repsFrom1;
+	struct dreplsrv_service *service = state->op->service;
+	struct dreplsrv_partition *partition = state->op->source_dsa->partition;
+	struct dreplsrv_drsuapi_connection *drsuapi = state->op->source_dsa->conn->drsuapi;
 	const struct drsuapi_DsReplicaOIDMapping_Ctr *mapping_ctr;
 	uint32_t object_count;
 	struct drsuapi_DsReplicaObjectListItemEx *first_object;
@@ -402,6 +425,7 @@ static void dreplsrv_op_pull_source_apply_changes_send(struct dreplsrv_op_pull_s
 	struct dsdb_extended_replicated_objects *objects;
 	bool more_data = false;
 	WERROR status;
+	NTSTATUS nt_status;
 
 	switch (ctr_level) {
 	case 1:
@@ -425,7 +449,8 @@ static void dreplsrv_op_pull_source_apply_changes_send(struct dreplsrv_op_pull_s
 		more_data			= ctr6->more_data;
 		break;
 	default:
-		composite_error(c, werror_to_ntstatus(WERR_BAD_NET_RESP));
+		nt_status = werror_to_ntstatus(WERR_BAD_NET_RESP);
+		tevent_req_nterror(req, nt_status);
 		return;
 	}
 
@@ -439,32 +464,39 @@ static void dreplsrv_op_pull_source_apply_changes_send(struct dreplsrv_op_pull_s
 							  &rf1,
 							  uptodateness_vector,
 							  &drsuapi->gensec_skey,
-							  st, &objects);
+							  state, &objects);
 	if (!W_ERROR_IS_OK(status)) {
-		DEBUG(0,("Failed to convert objects: %s\n", win_errstr(status)));
-		composite_error(c, werror_to_ntstatus(status));
+		nt_status = werror_to_ntstatus(WERR_BAD_NET_RESP);
+		DEBUG(0,("Failed to convert objects: %s/%s\n",
+			  win_errstr(status), nt_errstr(nt_status)));
+		tevent_req_nterror(req, nt_status);
 		return;
 	}
 
 	status = dsdb_extended_replicated_objects_commit(service->samdb,
 							 objects, 
-							 &st->op->source_dsa->notify_uSN);
+							 &state->op->source_dsa->notify_uSN);
 	talloc_free(objects);
 	if (!W_ERROR_IS_OK(status)) {
-		DEBUG(0,("Failed to commit objects: %s\n", win_errstr(status)));
-		composite_error(c, werror_to_ntstatus(status));
+		nt_status = werror_to_ntstatus(WERR_BAD_NET_RESP);
+		DEBUG(0,("Failed to commit objects: %s/%s\n",
+			  win_errstr(status), nt_errstr(nt_status)));
+		tevent_req_nterror(req, nt_status);
 		return;
 	}
 
 	/* if it applied fine, we need to update the highwatermark */
-	*st->op->source_dsa->repsFrom1 = rf1;
+	*state->op->source_dsa->repsFrom1 = rf1;
 
 	/*
 	 * TODO: update our uptodatevector!
 	 */
 
+	/* we don't need this maybe very large structure anymore */
+	TALLOC_FREE(r);
+
 	if (more_data) {
-		dreplsrv_op_pull_source_get_changes_send(st);
+		dreplsrv_op_pull_source_get_changes_trigger(req);
 		return;
 	}
 
@@ -473,43 +505,89 @@ static void dreplsrv_op_pull_source_apply_changes_send(struct dreplsrv_op_pull_s
 	   we join the domain, but they quickly expire.  We do it here
 	   so we can use the already established DRSUAPI pipe
 	*/
-	dreplsrv_update_refs_send(st);
+	dreplsrv_update_refs_trigger(req);
 }
 
-WERROR dreplsrv_op_pull_source_recv(struct composite_context *c)
+static void dreplsrv_update_refs_done(struct rpc_request *rreq);
+
+/*
+  send a UpdateRefs request to refresh our repsTo record on the server
+ */
+static void dreplsrv_update_refs_trigger(struct tevent_req *req)
 {
-	NTSTATUS status;
+	struct dreplsrv_op_pull_source_state *state = tevent_req_data(req,
+						      struct dreplsrv_op_pull_source_state);
+	struct dreplsrv_service *service = state->op->service;
+	struct dreplsrv_partition *partition = state->op->source_dsa->partition;
+	struct dreplsrv_drsuapi_connection *drsuapi = state->op->source_dsa->conn->drsuapi;
+	struct rpc_request *rreq;
+	struct drsuapi_DsReplicaUpdateRefs *r;
+	char *ntds_guid_str;
+	char *ntds_dns_name;
+
+	r = talloc(state, struct drsuapi_DsReplicaUpdateRefs);
+	if (tevent_req_nomem(r, req)) {
+		return;
+	}
 
-	status = composite_wait(c);
+	ntds_guid_str = GUID_string(r, &service->ntds_guid);
+	if (tevent_req_nomem(ntds_guid_str, req)) {
+		return;
+	}
 
-	talloc_free(c);
-	return ntstatus_to_werror(status);
+	ntds_dns_name = talloc_asprintf(r, "%s._msdcs.%s",
+					ntds_guid_str,
+					lp_dnsdomain(service->task->lp_ctx));
+	if (tevent_req_nomem(ntds_dns_name, req)) {
+		return;
+	}
+
+	r->in.bind_handle	= &drsuapi->bind_handle;
+	r->in.level             = 1;
+	r->in.req.req1.naming_context	  = &partition->nc;
+	r->in.req.req1.dest_dsa_dns_name  = ntds_dns_name;
+	r->in.req.req1.dest_dsa_guid	  = service->ntds_guid;
+	r->in.req.req1.options	          =
+		DRSUAPI_DS_REPLICA_UPDATE_ADD_REFERENCE |
+		DRSUAPI_DS_REPLICA_UPDATE_DELETE_REFERENCE;
+	if (!samdb_rodc(service->task->lp_ctx)) {
+		r->in.req.req1.options |= DRSUAPI_DS_REPLICA_UPDATE_WRITEABLE;
+	}
+
+	rreq = dcerpc_drsuapi_DsReplicaUpdateRefs_send(drsuapi->pipe, r, r);
+	if (tevent_req_nomem(rreq, req)) {
+		return;
+	}
+	composite_continue_rpc(NULL, rreq, dreplsrv_update_refs_done, req);
 }
 
 /*
   receive a UpdateRefs reply
  */
-static void dreplsrv_update_refs_recv(struct rpc_request *req)
+static void dreplsrv_update_refs_done(struct rpc_request *rreq)
 {
-	struct dreplsrv_op_pull_source_state *st = talloc_get_type(req->async.private_data,
-						   struct dreplsrv_op_pull_source_state);
-	struct composite_context *c = st->creq;
-	struct drsuapi_DsReplicaUpdateRefs *r = talloc_get_type(req->ndr.struct_ptr,
+	struct tevent_req *req = talloc_get_type(rreq->async.private_data,
+				 struct tevent_req);
+	struct drsuapi_DsReplicaUpdateRefs *r = talloc_get_type(rreq->ndr.struct_ptr,
 								struct drsuapi_DsReplicaUpdateRefs);
+	NTSTATUS status;
 
-	c->status = dcerpc_ndr_request_recv(req);
-	if (!composite_is_ok(c)) {
+	status = dcerpc_ndr_request_recv(rreq);
+	if (!NT_STATUS_IS_OK(status)) {
 		DEBUG(0,("UpdateRefs failed with %s\n", 
-			 nt_errstr(c->status)));
+			 nt_errstr(status)));
+		tevent_req_nterror(req, status);
 		return;
 	}
 
 	if (!W_ERROR_IS_OK(r->out.result)) {
-		DEBUG(0,("UpdateRefs failed with %s for %s %s\n", 
+		status = werror_to_ntstatus(r->out.result);
+		DEBUG(0,("UpdateRefs failed with %s/%s for %s %s\n",
 			 win_errstr(r->out.result),
+			 nt_errstr(status),
 			 r->in.req.req1.dest_dsa_dns_name,
 			 r->in.req.req1.naming_context->dn));
-		composite_error(c, werror_to_ntstatus(r->out.result));
+		tevent_req_nterror(req, status);
 		return;
 	}
 
@@ -517,46 +595,19 @@ static void dreplsrv_update_refs_recv(struct rpc_request *req)
 		 r->in.req.req1.dest_dsa_dns_name,
 		 r->in.req.req1.naming_context->dn));
 
-	composite_done(c);
+	tevent_req_done(req);
 }
 
-/*
-  send a UpdateRefs request to refresh our repsTo record on the server
- */
-static void dreplsrv_update_refs_send(struct dreplsrv_op_pull_source_state *st)
+WERROR dreplsrv_op_pull_source_recv(struct tevent_req *req)
 {
-	struct composite_context *c = st->creq;
-	struct dreplsrv_service *service = st->op->service;
-	struct dreplsrv_partition *partition = st->op->source_dsa->partition;
-	struct dreplsrv_drsuapi_connection *drsuapi = st->op->source_dsa->conn->drsuapi;
-	struct rpc_request *req;
-	struct drsuapi_DsReplicaUpdateRefs *r;
-	char *ntds_guid_str;
-	char *ntds_dns_name;
-
-	r = talloc(st, struct drsuapi_DsReplicaUpdateRefs);
-	if (composite_nomem(r, c)) return;
-
-	ntds_guid_str = GUID_string(r, &service->ntds_guid);
-	if (composite_nomem(ntds_guid_str, c)) return;
-
-	ntds_dns_name = talloc_asprintf(r, "%s._msdcs.%s",
-					ntds_guid_str,
-					lp_dnsdomain(service->task->lp_ctx));
-	if (composite_nomem(ntds_dns_name, c)) return;
+	NTSTATUS status;
 
-	r->in.bind_handle	= &drsuapi->bind_handle;
-	r->in.level             = 1;
-	r->in.req.req1.naming_context	  = &partition->nc;
-	r->in.req.req1.dest_dsa_dns_name  = ntds_dns_name;
-	r->in.req.req1.dest_dsa_guid	  = service->ntds_guid;
-	r->in.req.req1.options	          = 
-		DRSUAPI_DS_REPLICA_UPDATE_ADD_REFERENCE |
-		DRSUAPI_DS_REPLICA_UPDATE_DELETE_REFERENCE;
-	if (!samdb_rodc(service->task->lp_ctx)) {
-		r->in.req.req1.options |= DRSUAPI_DS_REPLICA_UPDATE_WRITEABLE;
+	if (tevent_req_is_nterror(req, &status)) {
+		tevent_req_received(req);
+		return ntstatus_to_werror(status);
 	}
 
-	req = dcerpc_drsuapi_DsReplicaUpdateRefs_send(drsuapi->pipe, r, r);
-	composite_continue_rpc(c, req, dreplsrv_update_refs_recv, st);
+	tevent_req_received(req);
+	return WERR_OK;
 }
+
diff --git a/source4/dsdb/repl/drepl_out_pull.c b/source4/dsdb/repl/drepl_out_pull.c
index 2793eec8b4..101214609a 100644
--- a/source4/dsdb/repl/drepl_out_pull.c
+++ b/source4/dsdb/repl/drepl_out_pull.c
@@ -33,21 +33,25 @@
 #include "librpc/gen_ndr/ndr_drsblobs.h"
 #include "libcli/composite/composite.h"
 
-static WERROR dreplsrv_schedule_partition_pull_source(struct dreplsrv_service *s,
-						      struct dreplsrv_partition *p,
-						      struct dreplsrv_partition_source_dsa *source,
-						      TALLOC_CTX *mem_ctx)
+WERROR dreplsrv_schedule_partition_pull_source(struct dreplsrv_service *s,
+					       struct dreplsrv_partition_source_dsa *source,
+					       enum drsuapi_DsExtendedOperation extended_op,
+					       uint64_t fsmo_info,
+					       dreplsrv_fsmo_callback_t callback)
 {
 	struct dreplsrv_out_operation *op;
 
-	op = talloc_zero(mem_ctx, struct dreplsrv_out_operation);
+	op = talloc_zero(s, struct dreplsrv_out_operation);
 	W_ERROR_HAVE_NO_MEMORY(op);
 
 	op->service	= s;
 	op->source_dsa	= source;
+	op->extended_op = extended_op;
+	op->fsmo_info   = fsmo_info;
+	op->callback    = callback;
 
 	DLIST_ADD_END(s->ops.pending, op, struct dreplsrv_out_operation *);
-	talloc_steal(s, op);
+
 	return WERR_OK;
 }
 
@@ -59,7 +63,7 @@ static WERROR dreplsrv_schedule_partition_pull(struct dreplsrv_service *s,
 	struct dreplsrv_partition_source_dsa *cur;
 
 	for (cur = p->sources; cur; cur = cur->next) {
-		status = dreplsrv_schedule_partition_pull_source(s, p, cur, mem_ctx);
+		status = dreplsrv_schedule_partition_pull_source(s, cur, DRSUAPI_EXOP_NONE, 0, NULL);
 		W_ERROR_NOT_OK_RETURN(status);
 	}
 
@@ -95,8 +99,10 @@ WERROR dreplsrv_schedule_partition_pull_by_guid(struct dreplsrv_service *s, TALL
 	return WERR_NOT_FOUND;
 }
 
-static void dreplsrv_pending_op_callback(struct dreplsrv_out_operation *op)
+static void dreplsrv_pending_op_callback(struct tevent_req *subreq)
 {
+	struct dreplsrv_out_operation *op = tevent_req_callback_data(subreq,
+					    struct dreplsrv_out_operation);
 	struct repsFromTo1 *rf = op->source_dsa->repsFrom1;
 	struct dreplsrv_service *s = op->service;
 	time_t t;
@@ -105,7 +111,8 @@ static void dreplsrv_pending_op_callback(struct dreplsrv_out_operation *op)
 	t = time(NULL);
 	unix_to_nt_time(&now, t);
 
-	rf->result_last_attempt = dreplsrv_op_pull_source_recv(op->creq);
+	rf->result_last_attempt = dreplsrv_op_pull_source_recv(subreq);
+	TALLOC_FREE(subreq);
 	if (W_ERROR_IS_OK(rf->result_last_attempt)) {
 		rf->consecutive_sync_failures	= 0;
 		rf->last_success		= now;
@@ -116,30 +123,28 @@ static void dreplsrv_pending_op_callback(struct dreplsrv_out_operation *op)
 
 	rf->consecutive_sync_failures++;
 
-	DEBUG(1,("dreplsrv_op_pull_source(%s/%s) failures[%u]\n",
-		win_errstr(rf->result_last_attempt),
-		nt_errstr(werror_to_ntstatus(rf->result_last_attempt)),
-		rf->consecutive_sync_failures));
+	DEBUG(1,("dreplsrv_op_pull_source(%s/%s) for %s failures[%u]\n",
+		 win_errstr(rf->result_last_attempt),
+		 win_errstr(rf->result_last_attempt),
+		 ldb_dn_get_linearized(op->source_dsa->partition->dn),
+		 rf->consecutive_sync_failures));
 
 done:
+	if (op->callback) {
+		op->callback(s, rf->result_last_attempt);
+	}
 	talloc_free(op);
 	s->ops.current = NULL;
 	dreplsrv_run_pending_ops(s);
 	dreplsrv_notify_run_ops(s);
 }
 
-static void dreplsrv_pending_op_callback_creq(struct composite_context *creq)
-{
-	struct dreplsrv_out_operation *op = talloc_get_type(creq->async.private_data,
-							   struct dreplsrv_out_operation);
-	dreplsrv_pending_op_callback(op);
-}
-
 void dreplsrv_run_pending_ops(struct dreplsrv_service *s)
 {
 	struct dreplsrv_out_operation *op;
 	time_t t;
 	NTTIME now;
+	struct tevent_req *subreq;
 
 	if (s->ops.current || s->ops.n_current) {
 		/* if there's still one running, we're done */
@@ -160,12 +165,18 @@ void dreplsrv_run_pending_ops(struct dreplsrv_service *s)
 
 	op->source_dsa->repsFrom1->last_attempt = now;
 
-	op->creq = dreplsrv_op_pull_source_send(op);
-	if (!op->creq) {
-		dreplsrv_pending_op_callback(op);
+	subreq = dreplsrv_op_pull_source_send(op, s->task->event_ctx, op);
+	if (!subreq) {
+		struct repsFromTo1 *rf = op->source_dsa->repsFrom1;
+
+		rf->result_last_attempt = WERR_NOMEM;
+		rf->consecutive_sync_failures++;
+
+		DEBUG(1,("dreplsrv_op_pull_source(%s/%s) failures[%u]\n",
+			win_errstr(rf->result_last_attempt),
+			nt_errstr(werror_to_ntstatus(rf->result_last_attempt)),
+			rf->consecutive_sync_failures));
 		return;
 	}
-
-	op->creq->async.fn		= dreplsrv_pending_op_callback_creq;
-	op->creq->async.private_data	= op;
+	tevent_req_set_callback(subreq, dreplsrv_pending_op_callback, op);
 }
diff --git a/source4/dsdb/repl/drepl_partitions.c b/source4/dsdb/repl/drepl_partitions.c
index 5b8227e7de..9a24fe541a 100644
--- a/source4/dsdb/repl/drepl_partitions.c
+++ b/source4/dsdb/repl/drepl_partitions.c
@@ -88,9 +88,9 @@ WERROR dreplsrv_load_partitions(struct dreplsrv_service *s)
 	return WERR_OK;
 }
 
-static WERROR dreplsrv_out_connection_attach(struct dreplsrv_service *s,
-					     const struct repsFromTo1 *rft,
-					     struct dreplsrv_out_connection **_conn)
+WERROR dreplsrv_out_connection_attach(struct dreplsrv_service *s,
+				      const struct repsFromTo1 *rft,
+				      struct dreplsrv_out_connection **_conn)
 {
 	struct dreplsrv_out_connection *cur, *conn = NULL;
 	const char *hostname;
@@ -188,6 +188,65 @@ 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)
+{
+	int 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;
+}
+
+/*
+  add our local UDV element for the partition
+ */
+static WERROR add_local_udv(struct dreplsrv_service *s,
+			    struct dreplsrv_partition *p,
+			    const struct GUID *our_invocation_id,
+			    struct drsuapi_DsReplicaCursorCtrEx *udv)
+{
+	int ret;
+	uint64_t highest_usn;
+	int i;
+
+	ret = dsdb_load_partition_usn(s->samdb, p->dn, &highest_usn);
+	if (ret != LDB_SUCCESS) {
+		/* nothing to add */
+		return WERR_OK;
+	}
+
+	for (i=0; i<udv->count; i++) {
+		if (GUID_equal(our_invocation_id, &udv->cursors[i].source_dsa_invocation_id)) {
+			udv->cursors[i].highest_usn = highest_usn;
+			return WERR_OK;
+		}
+	}
+
+	udv->cursors = talloc_realloc(p, udv->cursors, struct drsuapi_DsReplicaCursor, udv->count+1);
+	W_ERROR_HAVE_NO_MEMORY(udv->cursors);
+
+	udv->cursors[udv->count].source_dsa_invocation_id = *our_invocation_id;
+	udv->cursors[udv->count].highest_usn = highest_usn;
+	udv->count++;
+
+	return WERR_OK;
+}
+
 static WERROR dreplsrv_refresh_partition(struct dreplsrv_service *s,
 					 struct dreplsrv_partition *p)
 {
@@ -232,6 +291,11 @@ static WERROR dreplsrv_refresh_partition(struct dreplsrv_service *s,
 		talloc_free(nc_sid);
 	}
 
+	talloc_free(p->uptodatevector.cursors);
+	talloc_free(p->uptodatevector_ex.cursors);
+	ZERO_STRUCT(p->uptodatevector);
+	ZERO_STRUCT(p->uptodatevector_ex);
+
 	ouv_value = ldb_msg_find_ldb_val(r->msgs[0], "replUpToDateVector");
 	if (ouv_value) {
 		enum ndr_err_code ndr_err;
@@ -251,14 +315,14 @@ static WERROR dreplsrv_refresh_partition(struct dreplsrv_service *s,
 
 		p->uptodatevector.count		= ouv.ctr.ctr2.count;
 		p->uptodatevector.reserved	= ouv.ctr.ctr2.reserved;
-		talloc_free(p->uptodatevector.cursors);
 		p->uptodatevector.cursors	= talloc_steal(p, ouv.ctr.ctr2.cursors);
-	}
 
-	/*
-	 * TODO: add our own uptodatevector cursor
-	 */
+		status = udv_convert(p, &p->uptodatevector, &p->uptodatevector_ex);
+		W_ERROR_NOT_OK_RETURN(status);
+	}
 
+	status = add_local_udv(s, p, samdb_ntds_invocation_id(s->samdb), &p->uptodatevector_ex);
+	W_ERROR_NOT_OK_RETURN(status);
 
 	orf_el = ldb_msg_find_element(r->msgs[0], "repsFrom");
 	if (orf_el) {
diff --git a/source4/dsdb/repl/drepl_periodic.c b/source4/dsdb/repl/drepl_periodic.c
index 61d5598207..d2fbe45586 100644
--- a/source4/dsdb/repl/drepl_periodic.c
+++ b/source4/dsdb/repl/drepl_periodic.c
@@ -109,6 +109,8 @@ static void dreplsrv_periodic_run(struct dreplsrv_service *service)
 	/* the KCC might have changed repsFrom */
 	dreplsrv_refresh_partitions(service);
 
+	dreplsrv_ridalloc_check_rid_pool(service);
+
 	dreplsrv_run_pending_ops(service);
 	dreplsrv_notify_run_ops(service);
 }
diff --git a/source4/dsdb/repl/drepl_ridalloc.c b/source4/dsdb/repl/drepl_ridalloc.c
new file mode 100644
index 0000000000..43fc5a2c51
--- /dev/null
+++ b/source4/dsdb/repl/drepl_ridalloc.c
@@ -0,0 +1,282 @@
+/*
+   Unix SMB/CIFS mplementation.
+
+   DSDB replication service - RID allocation code
+
+   Copyright (C) Andrew Tridgell 2010
+   Copyright (C) Andrew Bartlett 2010
+
+   based on drepl_notify.c
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+*/
+
+#include "includes.h"
+#include "dsdb/samdb/samdb.h"
+#include "smbd/service.h"
+#include "dsdb/repl/drepl_service.h"
+#include "param/param.h"
+
+
+/*
+  create the RID manager source dsa structure
+ */
+static WERROR drepl_create_rid_manager_source_dsa(struct dreplsrv_service *service,
+						  struct ldb_dn *rid_manager_dn, struct ldb_dn *fsmo_role_dn)
+{
+	struct dreplsrv_partition_source_dsa *sdsa;
+	struct ldb_context *ldb = service->samdb;
+	int ret;
+	WERROR werr;
+
+	sdsa = talloc_zero(service, struct dreplsrv_partition_source_dsa);
+	W_ERROR_HAVE_NO_MEMORY(sdsa);
+
+	sdsa->partition = talloc_zero(sdsa, struct dreplsrv_partition);
+	if (!sdsa->partition) {
+		talloc_free(sdsa);
+		return WERR_NOMEM;
+	}
+
+	sdsa->partition->dn = samdb_base_dn(ldb);
+	sdsa->partition->nc.dn = ldb_dn_alloc_linearized(sdsa->partition, rid_manager_dn);
+	ret = dsdb_find_guid_by_dn(ldb, rid_manager_dn, &sdsa->partition->nc.guid);
+	if (ret != LDB_SUCCESS) {
+		DEBUG(0,(__location__ ": Failed to find GUID for %s\n",
+			 ldb_dn_get_linearized(rid_manager_dn)));
+		talloc_free(sdsa);
+		return WERR_DS_DRA_INTERNAL_ERROR;
+	}
+
+	sdsa->repsFrom1 = &sdsa->_repsFromBlob.ctr.ctr1;
+	ret = dsdb_find_guid_attr_by_dn(ldb, fsmo_role_dn, "objectGUID", &sdsa->repsFrom1->source_dsa_obj_guid);
+	if (ret != LDB_SUCCESS) {
+		DEBUG(0,(__location__ ": Failed to find objectGUID for %s\n",
+			 ldb_dn_get_linearized(fsmo_role_dn)));
+		talloc_free(sdsa);
+		return WERR_DS_DRA_INTERNAL_ERROR;
+	}
+
+	sdsa->repsFrom1->other_info = talloc_zero(sdsa, struct repsFromTo1OtherInfo);
+	if (!sdsa->repsFrom1->other_info) {
+		talloc_free(sdsa);
+		return WERR_NOMEM;
+	}
+
+	sdsa->repsFrom1->other_info->dns_name =
+		talloc_asprintf(sdsa->repsFrom1->other_info, "%s._msdcs.%s",
+				GUID_string(sdsa->repsFrom1->other_info, &sdsa->repsFrom1->source_dsa_obj_guid),
+				lp_dnsdomain(service->task->lp_ctx));
+	if (!sdsa->repsFrom1->other_info->dns_name) {
+		talloc_free(sdsa);
+		return WERR_NOMEM;
+	}
+
+
+	werr = dreplsrv_out_connection_attach(service, sdsa->repsFrom1, &sdsa->conn);
+	if (!W_ERROR_IS_OK(werr)) {
+		DEBUG(0,(__location__ ": Failed to attach to RID manager connection\n"));
+		talloc_free(sdsa);
+		return werr;
+	}
+
+	service->ridalloc.rid_manager_source_dsa = sdsa;
+	return WERR_OK;
+}
+
+/*
+  called when a rid allocation request has completed
+ */
+static void drepl_new_rid_pool_callback(struct dreplsrv_service *service, WERROR werr)
+{
+	if (!W_ERROR_IS_OK(werr)) {
+		DEBUG(0,(__location__ ": RID Manager failed RID allocation - %s\n",
+			 win_errstr(werr)));
+	} else {
+		DEBUG(3,(__location__ ": RID Manager completed RID allocation OK\n"));
+	}
+
+	/* don't keep the connection open to the RID Manager */
+	talloc_free(service->ridalloc.rid_manager_source_dsa);
+	service->ridalloc.rid_manager_source_dsa = NULL;
+
+	service->ridalloc.in_progress = false;
+}
+
+/*
+  schedule a getncchanges request to the RID Manager to ask for a new
+  set of RIDs using DRSUAPI_EXOP_FSMO_RID_ALLOC
+ */
+static WERROR drepl_request_new_rid_pool(struct dreplsrv_service *service,
+					 struct ldb_dn *rid_manager_dn, struct ldb_dn *fsmo_role_dn,
+					 uint64_t alloc_pool)
+{
+	WERROR werr;
+
+	if (service->ridalloc.rid_manager_source_dsa == NULL) {
+		/* we need to establish a connection to the RID
+		   Manager */
+		werr = drepl_create_rid_manager_source_dsa(service, rid_manager_dn, fsmo_role_dn);
+		W_ERROR_NOT_OK_RETURN(werr);
+	}
+
+	service->ridalloc.in_progress = true;
+
+	werr = dreplsrv_schedule_partition_pull_source(service, service->ridalloc.rid_manager_source_dsa,
+						       DRSUAPI_EXOP_FSMO_RID_ALLOC, alloc_pool,
+						       drepl_new_rid_pool_callback);
+	return werr;
+}
+
+
+/*
+  see if we are on the last pool we have
+ */
+static int drepl_ridalloc_pool_exhausted(struct ldb_context *ldb, bool *exhausted, uint64_t *alloc_pool)
+{
+	struct ldb_dn *server_dn, *machine_dn, *rid_set_dn;
+	TALLOC_CTX *tmp_ctx = talloc_new(ldb);
+	uint64_t prev_alloc_pool;
+	const char *attrs[] = { "rIDPreviousAllocationPool", "rIDAllocationPool", NULL };
+	int ret;
+	struct ldb_result *res;
+
+	server_dn = ldb_dn_get_parent(tmp_ctx, samdb_ntds_settings_dn(ldb));
+	if (!server_dn) {
+		talloc_free(tmp_ctx);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	ret = samdb_reference_dn(ldb, tmp_ctx, server_dn, "serverReference", &machine_dn);
+	if (ret != LDB_SUCCESS) {
+		DEBUG(0,(__location__ ": Failed to find serverReference in %s - %s",
+			 ldb_dn_get_linearized(server_dn), ldb_errstring(ldb)));
+		talloc_free(tmp_ctx);
+		return ret;
+	}
+
+	ret = samdb_reference_dn(ldb, tmp_ctx, machine_dn, "rIDSetReferences", &rid_set_dn);
+	if (ret == LDB_ERR_NO_SUCH_ATTRIBUTE) {
+		*exhausted = true;
+		talloc_free(tmp_ctx);
+		return LDB_SUCCESS;
+	}
+	if (ret != LDB_SUCCESS) {
+		DEBUG(0,(__location__ ": Failed to find rIDSetReferences in %s - %s",
+			 ldb_dn_get_linearized(machine_dn), ldb_errstring(ldb)));
+		talloc_free(tmp_ctx);
+		return ret;
+	}
+
+	ret = ldb_search(ldb, tmp_ctx, &res, rid_set_dn, LDB_SCOPE_BASE, attrs, NULL);
+	if (ret != LDB_SUCCESS) {
+		DEBUG(0,(__location__ ": Failed to load RID Set attrs from %s - %s",
+			 ldb_dn_get_linearized(rid_set_dn), ldb_errstring(ldb)));
+		talloc_free(tmp_ctx);
+		return ret;
+	}
+
+	*alloc_pool = ldb_msg_find_attr_as_uint64(res->msgs[0], "rIDAllocationPool", 0);
+	prev_alloc_pool = ldb_msg_find_attr_as_uint64(res->msgs[0], "rIDPreviousAllocationPool", 0);
+
+	if (*alloc_pool != prev_alloc_pool) {
+		*exhausted = false;
+	} else {
+		*exhausted = true;
+	}
+
+	talloc_free(tmp_ctx);
+	return LDB_SUCCESS;
+}
+
+
+/*
+  see if we are low on RIDs in the RID Set rIDAllocationPool. If we
+  are, then schedule a replication call with DRSUAPI_EXOP_FSMO_RID_ALLOC
+  to the RID Manager
+ */
+WERROR dreplsrv_ridalloc_check_rid_pool(struct dreplsrv_service *service)
+{
+	struct ldb_dn *rid_manager_dn, *fsmo_role_dn;
+	TALLOC_CTX *tmp_ctx = talloc_new(service);
+	struct ldb_context *ldb = service->samdb;
+	bool exhausted;
+	WERROR werr;
+	int ret;
+	uint64_t alloc_pool;
+
+	if (service->ridalloc.in_progress) {
+		talloc_free(tmp_ctx);
+		return WERR_OK;
+	}
+
+	/*
+	  steps:
+	    - find who the RID Manager is
+	    - if we are the RID Manager then nothing to do
+	    - find our RID Set object
+	    - load rIDAllocationPool and rIDPreviousAllocationPool
+	    - if rIDAllocationPool != rIDPreviousAllocationPool then
+	      nothing to do
+	    - schedule a getncchanges with DRSUAPI_EXOP_FSMO_RID_ALLOC
+	      to the RID Manager
+	 */
+
+	/* work out who is the RID Manager */
+	ret = samdb_rid_manager_dn(ldb, tmp_ctx, &rid_manager_dn);
+	if (ret != LDB_SUCCESS) {
+		DEBUG(0, (__location__ ": Failed to find RID Manager object - %s", ldb_errstring(ldb)));
+		talloc_free(tmp_ctx);
+		return WERR_DS_DRA_INTERNAL_ERROR;
+	}
+
+	/* find the DN of the RID Manager */
+	ret = samdb_reference_dn(ldb, tmp_ctx, rid_manager_dn, "fSMORoleOwner", &fsmo_role_dn);
+	if (ret != LDB_SUCCESS) {
+		DEBUG(0,(__location__ ": Failed to find fSMORoleOwner in RID Manager object - %s",
+			 ldb_errstring(ldb)));
+		talloc_free(tmp_ctx);
+		return WERR_DS_DRA_INTERNAL_ERROR;
+	}
+
+	if (ldb_dn_compare(samdb_ntds_settings_dn(ldb), fsmo_role_dn) == 0) {
+		/* we are the RID Manager - no need to do a
+		   DRSUAPI_EXOP_FSMO_RID_ALLOC */
+		talloc_free(tmp_ctx);
+		return WERR_OK;
+	}
+
+	ret = drepl_ridalloc_pool_exhausted(ldb, &exhausted, &alloc_pool);
+	if (ret != LDB_SUCCESS) {
+		talloc_free(tmp_ctx);
+		return WERR_DS_DRA_INTERNAL_ERROR;
+	}
+
+	DEBUG(2,(__location__ ": Requesting more RIDs from RID Manager\n"));
+
+	werr = drepl_request_new_rid_pool(service, rid_manager_dn, fsmo_role_dn, alloc_pool);
+	talloc_free(tmp_ctx);
+	return werr;
+}
+
+/* called by the samldb ldb module to tell us to ask for a new RID
+   pool */
+void dreplsrv_allocate_rid(struct messaging_context *msg, void *private_data,
+			   uint32_t msg_type,
+			   struct server_id server_id, DATA_BLOB *data)
+{
+	struct dreplsrv_service *service = talloc_get_type(private_data, struct dreplsrv_service);
+	dreplsrv_ridalloc_check_rid_pool(service);
+}
diff --git a/source4/dsdb/repl/drepl_service.c b/source4/dsdb/repl/drepl_service.c
index a05ccc8d70..44164ff68f 100644
--- a/source4/dsdb/repl/drepl_service.c
+++ b/source4/dsdb/repl/drepl_service.c
@@ -206,6 +206,7 @@ static void dreplsrv_task_init(struct task_server *task)
 	irpc_add_name(task->msg_ctx, "dreplsrv");
 
 	IRPC_REGISTER(task->msg_ctx, drsuapi, DRSUAPI_DSREPLICASYNC, drepl_replica_sync, service);
+	messaging_register(task->msg_ctx, service, MSG_DREPL_ALLOCATE_RID, dreplsrv_allocate_rid);
 }
 
 /*
diff --git a/source4/dsdb/repl/drepl_service.h b/source4/dsdb/repl/drepl_service.h
index 0f9684fa78..0a0d721d5c 100644
--- a/source4/dsdb/repl/drepl_service.h
+++ b/source4/dsdb/repl/drepl_service.h
@@ -100,6 +100,8 @@ struct dreplsrv_partition {
 	struct dreplsrv_partition_source_dsa *sources;
 };
 
+typedef void (*dreplsrv_fsmo_callback_t)(struct dreplsrv_service *, WERROR );
+
 struct dreplsrv_out_operation {
 	struct dreplsrv_out_operation *prev, *next;
 
@@ -107,7 +109,9 @@ struct dreplsrv_out_operation {
 
 	struct dreplsrv_partition_source_dsa *source_dsa;
 
-	struct composite_context *creq;
+	enum drsuapi_DsExtendedOperation extended_op;
+	uint64_t fsmo_info;
+	dreplsrv_fsmo_callback_t callback;
 };
 
 struct dreplsrv_notify_operation {
@@ -204,6 +208,11 @@ struct dreplsrv_service {
 		/* an active notify operation */
 		struct dreplsrv_notify_operation *n_current;
 	} ops;
+
+	struct {
+		bool in_progress;
+		struct dreplsrv_partition_source_dsa *rid_manager_source_dsa;
+	} ridalloc;
 };
 
 #include "dsdb/repl/drepl_out_helpers.h"