DEBUG ridalloc...
[metze/samba/wip.git] / source4 / dsdb / repl / drepl_out_helpers.c
index c292c6db744e4001f3ddcb834bf1ebf69287c56e..1ea519d16abdcc3b0e3481bd8d206bc7c1bd99c2 100644 (file)
 #include "librpc/gen_ndr/ndr_drsblobs.h"
 #include "libcli/composite/composite.h"
 #include "auth/gensec/gensec.h"
+#include "param/param.h"
+#include "../lib/util/tevent_ntstatus.h"
 
 struct dreplsrv_out_drsuapi_state {
-       struct composite_context *creq;
+       struct tevent_context *ev;
 
        struct dreplsrv_out_connection *conn;
 
@@ -45,213 +47,259 @@ struct dreplsrv_out_drsuapi_state {
        struct drsuapi_DsBind bind_r;
 };
 
-static void dreplsrv_out_drsuapi_connect_recv(struct composite_context *creq);
+static void dreplsrv_out_drsuapi_connect_done(struct composite_context *creq);
 
-static struct composite_context *dreplsrv_out_drsuapi_send(struct dreplsrv_out_connection *conn)
+struct tevent_req *dreplsrv_out_drsuapi_send(TALLOC_CTX *mem_ctx,
+                                            struct tevent_context *ev,
+                                            struct dreplsrv_out_connection *conn)
 {
-       struct composite_context *c;
+       struct tevent_req *req;
+       struct dreplsrv_out_drsuapi_state *state;
        struct composite_context *creq;
-       struct dreplsrv_out_drsuapi_state *st;
-
-       c = composite_create(conn, conn->service->task->event_ctx);
-       if (c == NULL) return NULL;
 
-       st = talloc_zero(c, struct dreplsrv_out_drsuapi_state);
-       if (composite_nomem(st, c)) return c;
+       req = tevent_req_create(mem_ctx, &state,
+                               struct dreplsrv_out_drsuapi_state);
+       if (req == NULL) {
+               return NULL;
+       }
 
-       c->private_data = st;
+       state->ev       = ev;
+       state->conn     = conn;
+       state->drsuapi  = conn->drsuapi;
 
-       st->creq        = c;
-       st->conn        = conn;
-       st->drsuapi     = conn->drsuapi;
+       if (state->drsuapi && !state->drsuapi->pipe->conn->dead) {
+               tevent_req_done(req);
+               return tevent_req_post(req, ev);
+       }
 
-       if (st->drsuapi && !st->drsuapi->pipe->conn->dead) {
-               composite_done(c);
-               return c;
-       } else if (st->drsuapi && st->drsuapi->pipe->conn->dead) {
-               talloc_free(st->drsuapi);
+       if (state->drsuapi && state->drsuapi->pipe->conn->dead) {
+               talloc_free(state->drsuapi);
                conn->drsuapi = NULL;
        }
 
-       st->drsuapi     = talloc_zero(st, struct dreplsrv_drsuapi_connection);
-       if (composite_nomem(st->drsuapi, c)) return c;
+       state->drsuapi = talloc_zero(state, struct dreplsrv_drsuapi_connection);
+       if (tevent_req_nomem(state->drsuapi, req)) {
+               return tevent_req_post(req, ev);
+       }
 
-       creq = dcerpc_pipe_connect_b_send(st, conn->binding, &ndr_table_drsuapi,
+       creq = dcerpc_pipe_connect_b_send(state, conn->binding, &ndr_table_drsuapi,
                                          conn->service->system_session_info->credentials,
-                                         c->event_ctx, conn->service->task->lp_ctx);
-       composite_continue(c, creq, dreplsrv_out_drsuapi_connect_recv, st);
+                                         ev, conn->service->task->lp_ctx);
+       if (tevent_req_nomem(creq, req)) {
+               return tevent_req_post(req, ev);
+       }
+       composite_continue(NULL, creq, dreplsrv_out_drsuapi_connect_done, req);
 
-       return c;
+       return req;
 }
 
-static void dreplsrv_out_drsuapi_bind_send(struct dreplsrv_out_drsuapi_state *st);
+static void dreplsrv_out_drsuapi_bind_done(struct tevent_req *subreq);
 
-static void dreplsrv_out_drsuapi_connect_recv(struct composite_context *creq)
+static void dreplsrv_out_drsuapi_connect_done(struct composite_context *creq)
 {
-       struct dreplsrv_out_drsuapi_state *st = talloc_get_type(creq->async.private_data,
-                                               struct dreplsrv_out_drsuapi_state);
-       struct composite_context *c = st->creq;
-
-       c->status = dcerpc_pipe_connect_b_recv(creq, st->drsuapi, &st->drsuapi->pipe);
-       if (!composite_is_ok(c)) return;
-
-       c->status = gensec_session_key(st->drsuapi->pipe->conn->security_state.generic_state,
-                                      &st->drsuapi->gensec_skey);
-       if (!composite_is_ok(c)) return;
+       struct tevent_req *req = talloc_get_type(creq->async.private_data,
+                                                struct tevent_req);
+       struct dreplsrv_out_drsuapi_state *state = tevent_req_data(req,
+                                                  struct dreplsrv_out_drsuapi_state);
+       NTSTATUS status;
+       struct tevent_req *subreq;
 
-       dreplsrv_out_drsuapi_bind_send(st);
-}
+       status = dcerpc_pipe_connect_b_recv(creq,
+                                           state->drsuapi,
+                                           &state->drsuapi->pipe);
+       if (tevent_req_nterror(req, status)) {
+               return;
+       }
 
-static void dreplsrv_out_drsuapi_bind_recv(struct rpc_request *req);
+       state->drsuapi->drsuapi_handle = state->drsuapi->pipe->binding_handle;
 
-static void dreplsrv_out_drsuapi_bind_send(struct dreplsrv_out_drsuapi_state *st)
-{
-       struct composite_context *c = st->creq;
-       struct rpc_request *req;
+       status = gensec_session_key(state->drsuapi->pipe->conn->security_state.generic_state,
+                                   &state->drsuapi->gensec_skey);
+       if (tevent_req_nterror(req, status)) {
+               return;
+       }
 
-       st->bind_info_ctr.length        = 28;
-       st->bind_info_ctr.info.info28   = st->conn->service->bind_info28;
+       state->bind_info_ctr.length             = 28;
+       state->bind_info_ctr.info.info28        = state->conn->service->bind_info28;
 
-       st->bind_r.in.bind_guid = &st->conn->service->ntds_guid;
-       st->bind_r.in.bind_info = &st->bind_info_ctr;
-       st->bind_r.out.bind_handle = &st->drsuapi->bind_handle;
+       state->bind_r.in.bind_guid = &state->conn->service->ntds_guid;
+       state->bind_r.in.bind_info = &state->bind_info_ctr;
+       state->bind_r.out.bind_handle = &state->drsuapi->bind_handle;
 
-       req = dcerpc_drsuapi_DsBind_send(st->drsuapi->pipe, st, &st->bind_r);
-       composite_continue_rpc(c, req, dreplsrv_out_drsuapi_bind_recv, st);
+       subreq = dcerpc_drsuapi_DsBind_r_send(state,
+                                             state->ev,
+                                             state->drsuapi->drsuapi_handle,
+                                             &state->bind_r);
+       if (tevent_req_nomem(subreq, req)) {
+               return;
+       }
+       tevent_req_set_callback(subreq, dreplsrv_out_drsuapi_bind_done, req);
 }
 
-static void dreplsrv_out_drsuapi_bind_recv(struct rpc_request *req)
+static void dreplsrv_out_drsuapi_bind_done(struct tevent_req *subreq)
 {
-       struct dreplsrv_out_drsuapi_state *st = talloc_get_type(req->async.private_data,
-                                               struct dreplsrv_out_drsuapi_state);
-       struct composite_context *c = st->creq;
+       struct tevent_req *req = tevent_req_callback_data(subreq,
+                                struct tevent_req);
+       struct dreplsrv_out_drsuapi_state *state = tevent_req_data(req,
+                                                  struct dreplsrv_out_drsuapi_state);
+       NTSTATUS status;
 
-       c->status = dcerpc_ndr_request_recv(req);
-       if (!composite_is_ok(c)) return;
+       status = dcerpc_drsuapi_DsBind_r_recv(subreq, state);
+       TALLOC_FREE(subreq);
+       if (tevent_req_nterror(req, status)) {
+               return;
+       }
 
-       if (!W_ERROR_IS_OK(st->bind_r.out.result)) {
-               composite_error(c, werror_to_ntstatus(st->bind_r.out.result));
+       if (!W_ERROR_IS_OK(state->bind_r.out.result)) {
+               status = werror_to_ntstatus(state->bind_r.out.result);
+               tevent_req_nterror(req, status);
                return;
        }
 
-       ZERO_STRUCT(st->drsuapi->remote_info28);
-       if (st->bind_r.out.bind_info) {
-               switch (st->bind_r.out.bind_info->length) {
+       ZERO_STRUCT(state->drsuapi->remote_info28);
+       if (state->bind_r.out.bind_info) {
+               struct drsuapi_DsBindInfo28 *info28;
+               info28 = &state->drsuapi->remote_info28;
+
+               switch (state->bind_r.out.bind_info->length) {
                case 24: {
                        struct drsuapi_DsBindInfo24 *info24;
-                       info24 = &st->bind_r.out.bind_info->info.info24;
-                       st->drsuapi->remote_info28.supported_extensions = info24->supported_extensions;
-                       st->drsuapi->remote_info28.site_guid            = info24->site_guid;
-                       st->drsuapi->remote_info28.pid                  = info24->pid;
-                       st->drsuapi->remote_info28.repl_epoch           = 0;
+                       info24 = &state->bind_r.out.bind_info->info.info24;
+
+                       info28->supported_extensions    = info24->supported_extensions;
+                       info28->site_guid               = info24->site_guid;
+                       info28->pid                     = info24->pid;
+                       info28->repl_epoch              = 0;
                        break;
                }
                case 48: {
                        struct drsuapi_DsBindInfo48 *info48;
-                       info48 = &st->bind_r.out.bind_info->info.info48;
-                       st->drsuapi->remote_info28.supported_extensions = info48->supported_extensions;
-                       st->drsuapi->remote_info28.site_guid            = info48->site_guid;
-                       st->drsuapi->remote_info28.pid                  = info48->pid;
-                       st->drsuapi->remote_info28.repl_epoch           = info48->repl_epoch;
+                       info48 = &state->bind_r.out.bind_info->info.info48;
+
+                       info28->supported_extensions    = info48->supported_extensions;
+                       info28->site_guid               = info48->site_guid;
+                       info28->pid                     = info48->pid;
+                       info28->repl_epoch              = info48->repl_epoch;
                        break;
                }
                case 28:
-                       st->drsuapi->remote_info28 = st->bind_r.out.bind_info->info.info28;
+                       *info28 = state->bind_r.out.bind_info->info.info28;
                        break;
                }
        }
 
-       composite_done(c);
+       tevent_req_done(req);
 }
 
-static NTSTATUS dreplsrv_out_drsuapi_recv(struct composite_context *c)
+NTSTATUS dreplsrv_out_drsuapi_recv(struct tevent_req *req)
 {
+       struct dreplsrv_out_drsuapi_state *state = tevent_req_data(req,
+                                                  struct dreplsrv_out_drsuapi_state);
        NTSTATUS status;
-       struct dreplsrv_out_drsuapi_state *st = talloc_get_type(c->private_data,
-                                               struct dreplsrv_out_drsuapi_state);
 
-       status = composite_wait(c);
-
-       if (NT_STATUS_IS_OK(status)) {
-               st->conn->drsuapi = talloc_steal(st->conn, st->drsuapi);
+       if (tevent_req_is_nterror(req, &status)) {
+               tevent_req_received(req);
+               return status;
        }
 
-       talloc_free(c);
-       return status;
+       state->conn->drsuapi = talloc_move(state->conn, &state->drsuapi);
+
+       tevent_req_received(req);
+       return NT_STATUS_OK;
 }
 
 struct dreplsrv_op_pull_source_state {
-       struct composite_context *creq;
-
+       struct tevent_context *ev;
        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;
+       void *ndr_struct_ptr;
 };
 
-static void dreplsrv_op_pull_source_connect_recv(struct composite_context *creq);
+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 composite_context *creq;
-       struct dreplsrv_op_pull_source_state *st;
-
-       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;
-
-       st->creq        = c;
-       st->op          = op;
+       struct tevent_req *req;
+       struct dreplsrv_op_pull_source_state *state;
+       struct tevent_req *subreq;
+
+       req = tevent_req_create(mem_ctx, &state,
+                               struct dreplsrv_op_pull_source_state);
+       if (req == NULL) {
+               return NULL;
+       }
+       state->ev = ev;
+       state->op = op;
 
-       creq = dreplsrv_out_drsuapi_send(op->source_dsa->conn);
-       composite_continue(c, creq, dreplsrv_op_pull_source_connect_recv, 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_recv(struct composite_context *creq)
+static void dreplsrv_op_pull_source_connect_done(struct tevent_req *subreq)
 {
-       struct dreplsrv_op_pull_source_state *st = talloc_get_type(creq->async.private_data,
-                                                  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(creq);
-       if (!composite_is_ok(c)) return;
+       status = dreplsrv_out_drsuapi_recv(subreq);
+       TALLOC_FREE(subreq);
+       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 tevent_req *subreq);
 
-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 drsuapi_DsGetNCChanges *r;
+       struct drsuapi_DsReplicaCursorCtrEx *uptodateness_vector;
+       struct tevent_req *subreq;
 
-       r = talloc(st, struct drsuapi_DsGetNCChanges);
-       if (composite_nomem(r, c)) return;
+       if ((rf1->replica_flags & DRSUAPI_DRS_WRIT_REP) == 0 &&
+           state->op->extended_op == DRSUAPI_EXOP_NONE) {
+               return;
+       }
 
-       r->out.level_out = talloc(r, int32_t);
-       if (composite_nomem(r->out.level_out, c)) return;
+       r = talloc(state, struct drsuapi_DsGetNCChanges);
+       if (tevent_req_nomem(r, req)) {
+               return;
+       }
+
+       r->out.level_out = talloc(r, uint32_t);
+       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) {
@@ -260,12 +308,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;
@@ -276,40 +324,61 @@ 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 1
+       NDR_PRINT_IN_DEBUG(drsuapi_DsGetNCChanges, r);
+#endif
+
+       state->ndr_struct_ptr = r;
+       subreq = dcerpc_drsuapi_DsGetNCChanges_r_send(state,
+                                                     state->ev,
+                                                     drsuapi->drsuapi_handle,
+                                                     r);
+       if (tevent_req_nomem(subreq, req)) {
+               return;
+       }
+       tevent_req_set_callback(subreq, 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 tevent_req *subreq)
 {
-       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 = tevent_req_callback_data(subreq,
+                                struct tevent_req);
+       struct dreplsrv_op_pull_source_state *state = tevent_req_data(req,
+                                                     struct dreplsrv_op_pull_source_state);
+       NTSTATUS status;
+       struct drsuapi_DsGetNCChanges *r = talloc_get_type(state->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;
+       state->ndr_struct_ptr = NULL;
+
+       status = dcerpc_drsuapi_DsGetNCChanges_r_recv(subreq, r);
+       TALLOC_FREE(subreq);
+       if (tevent_req_nterror(req, status)) {
+               return;
+       }
+
+       NDR_PRINT_OUT_DEBUG(drsuapi_DsGetNCChanges, r);
 
        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;
        }
 
@@ -336,44 +405,52 @@ 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_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_update_refs_trigger(struct tevent_req *req);
+
+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;
        uint32_t linked_attributes_count;
        struct drsuapi_DsReplicaLinkedAttribute *linked_attributes;
        const struct drsuapi_DsReplicaCursor2CtrEx *uptodateness_vector;
+       struct dsdb_extended_replicated_objects *objects;
        bool more_data = false;
        WERROR status;
+       NTSTATUS nt_status;
 
        switch (ctr_level) {
        case 1:
@@ -397,48 +474,179 @@ 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;
+       }
+
+       status = dsdb_extended_replicated_objects_convert(service->samdb,
+                                                         partition->nc.dn,
+                                                         mapping_ctr,
+                                                         object_count,
+                                                         first_object,
+                                                         linked_attributes_count,
+                                                         linked_attributes,
+                                                         &rf1,
+                                                         uptodateness_vector,
+                                                         &drsuapi->gensec_skey,
+                                                         state, &objects);
+       if (!W_ERROR_IS_OK(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,
-                                                        partition->nc.dn,
-                                                        mapping_ctr,
-                                                        object_count,
-                                                        first_object,
-                                                        linked_attributes_count,
-                                                        linked_attributes,
-                                                        &rf1,
-                                                        uptodateness_vector,
-                                                        &drsuapi->gensec_skey,
-                                                        st, NULL);
+                                                        objects, 
+                                                        &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;
+       }
+
+       /* now we need to update the repsTo record for this partition
+          on the server. These records are initially established when
+          we join the domain, but they quickly expire.  We do it here
+          so we can use the already established DRSUAPI pipe
+       */
+       if (state->op->extended_op == DRSUAPI_EXOP_NONE) {
+               dreplsrv_update_refs_trigger(req);
+       } else {
+               tevent_req_done(req);
+       }
+}
+
+static void dreplsrv_update_refs_done(struct tevent_req *subreq);
+
+/*
+  send a UpdateRefs request to refresh our repsTo record on the server
+ */
+static void dreplsrv_update_refs_trigger(struct tevent_req *req)
+{
+       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 drsuapi_DsReplicaUpdateRefs *r;
+       char *ntds_guid_str;
+       char *ntds_dns_name;
+       struct tevent_req *subreq;
+       bool am_rodc;
+       int ret;
+
+       r = talloc(state, struct drsuapi_DsReplicaUpdateRefs);
+       if (tevent_req_nomem(r, req)) {
+               return;
+       }
+
+       ntds_guid_str = GUID_string(r, &service->ntds_guid);
+       if (tevent_req_nomem(ntds_guid_str, req)) {
+               return;
+       }
+
+       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_DRS_ADD_REF | DRSUAPI_DRS_DEL_REF;
+       ret = samdb_rodc(service->samdb, &am_rodc);
+       if (ret == LDB_SUCCESS && !am_rodc) {
+               r->in.req.req1.options |= DRSUAPI_DRS_WRIT_REP;
+       }
+
+       state->ndr_struct_ptr = r;
+       subreq = dcerpc_drsuapi_DsReplicaUpdateRefs_r_send(state,
+                                                          state->ev,
+                                                          drsuapi->drsuapi_handle,
+                                                          r);
+       if (tevent_req_nomem(subreq, req)) {
+               return;
+       }
+       tevent_req_set_callback(subreq, dreplsrv_update_refs_done, req);
+}
+
+/*
+  receive a UpdateRefs reply
+ */
+static void dreplsrv_update_refs_done(struct tevent_req *subreq)
+{
+       struct tevent_req *req = tevent_req_callback_data(subreq,
+                                struct tevent_req);
+       struct dreplsrv_op_pull_source_state *state = tevent_req_data(req,
+                                                     struct dreplsrv_op_pull_source_state);
+       struct drsuapi_DsReplicaUpdateRefs *r = talloc_get_type(state->ndr_struct_ptr,
+                                                               struct drsuapi_DsReplicaUpdateRefs);
+       NTSTATUS status;
+
+       state->ndr_struct_ptr = NULL;
+
+       status = dcerpc_drsuapi_DsReplicaUpdateRefs_r_recv(subreq, r);
+       TALLOC_FREE(subreq);
+       if (!NT_STATUS_IS_OK(status)) {
+               DEBUG(0,("UpdateRefs failed with %s\n", 
+                        nt_errstr(status)));
+               tevent_req_nterror(req, status);
                return;
        }
 
-       composite_done(c);
+       if (!W_ERROR_IS_OK(r->out.result)) {
+               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));
+               tevent_req_nterror(req, status);
+               return;
+       }
+
+       DEBUG(4,("UpdateRefs OK for %s %s\n", 
+                r->in.req.req1.dest_dsa_dns_name,
+                r->in.req.req1.naming_context->dn));
+
+       tevent_req_done(req);
 }
 
-WERROR dreplsrv_op_pull_source_recv(struct composite_context *c)
+WERROR dreplsrv_op_pull_source_recv(struct tevent_req *req)
 {
        NTSTATUS status;
 
-       status = composite_wait(c);
+       if (tevent_req_is_nterror(req, &status)) {
+               tevent_req_received(req);
+               return ntstatus_to_werror(status);
+       }
 
-       talloc_free(c);
-       return ntstatus_to_werror(status);
+       tevent_req_received(req);
+       return WERR_OK;
 }
+