Changeset 740 for vendor/current/source4/dsdb/repl

Timestamp:

Nov 14, 2012, 12:59:34 PM (13 years ago)

Author:

Silvan Scherrer

Message:

Samba Server: update vendor to 3.6.0

Location:

vendor/current/source4/dsdb/repl

Files:

• drepl_extended.c (added)
• drepl_fsmo.c (added)
• drepl_notify.c (modified) (14 diffs)
• drepl_out_helpers.c (modified) (11 diffs)
• drepl_out_pull.c (modified) (5 diffs)
• drepl_partitions.c (modified) (12 diffs)
• drepl_periodic.c (modified) (4 diffs)
• drepl_replica.c (added)
• drepl_ridalloc.c (added)
• drepl_secret.c (added)
• drepl_service.c (modified) (9 diffs)
• drepl_service.h (modified) (6 diffs)
• replicated_objects.c (modified) (18 diffs)

vendor/current/source4/dsdb/repl/drepl_notify.c

@@ -27 +27 @@
 #include "auth/auth.h"
 #include "smbd/service.h"
-#include "lib/messaging/irpc.h"
 #include "dsdb/repl/drepl_service.h"
-#include "lib/ldb/include/ldb_errors.h"
+#include <ldb_errors.h>
 #include "../lib/util/dlinklist.h"
 #include "librpc/gen_ndr/ndr_misc.h"
@@ -35 +34 @@
 #include "librpc/gen_ndr/ndr_drsblobs.h"
 #include "libcli/composite/composite.h"
+#include "../lib/util/tevent_ntstatus.h"
 
 
 struct dreplsrv_op_notify_state {
-        struct composite_context *creq;
-
-        struct dreplsrv_out_connection *conn;
-
-        struct dreplsrv_drsuapi_connection *drsuapi;
-
-        struct drsuapi_DsBindInfoCtr bind_info_ctr;
-        struct drsuapi_DsBind bind_r;
+        struct tevent_context *ev;
         struct dreplsrv_notify_operation *op;
+        void *ndr_struct_ptr;
 };
 
-/*
-  receive a DsReplicaSync reply
- */
-static void dreplsrv_op_notify_replica_sync_recv(struct rpc_request *req)
-{
-        struct dreplsrv_op_notify_state *st = talloc_get_type(req->async.private_data,
-                                                              struct dreplsrv_op_notify_state);
-        struct composite_context *c = st->creq;
-        struct drsuapi_DsReplicaSync *r = talloc_get_type(req->ndr.struct_ptr,
-                                                          struct drsuapi_DsReplicaSync);
-
-        c->status = dcerpc_ndr_request_recv(req);
-        if (!composite_is_ok(c)) return;
-
-        if (!W_ERROR_IS_OK(r->out.result)) {
-                composite_error(c, werror_to_ntstatus(r->out.result));
-                return;
-        }
-
-        composite_done(c);
-}
-
-/*
-  send a DsReplicaSync
-*/
-static void dreplsrv_op_notify_replica_sync_send(struct dreplsrv_op_notify_state *st)
-{
-        struct composite_context *c = st->creq;
-        struct dreplsrv_partition *partition = st->op->source_dsa->partition;
-        struct dreplsrv_drsuapi_connection *drsuapi = st->op->source_dsa->conn->drsuapi;
-        struct rpc_request *req;
+static void dreplsrv_op_notify_connect_done(struct tevent_req *subreq);
+
+/*
+  start the ReplicaSync async call
+ */
+static struct tevent_req *dreplsrv_op_notify_send(TALLOC_CTX *mem_ctx,
+                                                  struct tevent_context *ev,
+                                                  struct dreplsrv_notify_operation *op)
+{
+        struct tevent_req *req;
+        struct dreplsrv_op_notify_state *state;
+        struct tevent_req *subreq;
+
+        req = tevent_req_create(mem_ctx, &state,
+                                struct dreplsrv_op_notify_state);
+        if (req == NULL) {
+                return NULL;
+        }
+        state->ev = ev;
+        state->op = op;
+
+        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_notify_connect_done, req);
+
+        return req;
+}
+
+static void dreplsrv_op_notify_replica_sync_trigger(struct tevent_req *req);
+
+static void dreplsrv_op_notify_connect_done(struct tevent_req *subreq)
+{
+        struct tevent_req *req = tevent_req_callback_data(subreq,
+                                                          struct tevent_req);
+        NTSTATUS status;
+
+        status = dreplsrv_out_drsuapi_recv(subreq);
+        TALLOC_FREE(subreq);
+        if (tevent_req_nterror(req, status)) {
+                return;
+        }
+
+        dreplsrv_op_notify_replica_sync_trigger(req);
+}
+
+static void dreplsrv_op_notify_replica_sync_done(struct tevent_req *subreq);
+
+static void dreplsrv_op_notify_replica_sync_trigger(struct tevent_req *req)
+{
+        struct dreplsrv_op_notify_state *state =
+                tevent_req_data(req,
+                struct dreplsrv_op_notify_state);
+        struct dreplsrv_partition *partition = state->op->source_dsa->partition;
+        struct dreplsrv_drsuapi_connection *drsuapi = state->op->source_dsa->conn->drsuapi;
         struct drsuapi_DsReplicaSync *r;
-
-        r = talloc_zero(st, struct drsuapi_DsReplicaSync);
-        if (composite_nomem(r, c)) return;
-
+        struct tevent_req *subreq;
+
+        r = talloc_zero(state, struct drsuapi_DsReplicaSync);
+        if (tevent_req_nomem(r, req)) {
+                return;
+        }
+        r->in.req = talloc_zero(r, union drsuapi_DsReplicaSyncRequest);
+        if (tevent_req_nomem(r, 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.source_dsa_guid = st->op->service->ntds_guid;
-        r->in.req.req1.options = 
-                DRSUAPI_DS_REPLICA_SYNC_ASYNCHRONOUS_OPERATION |
-                DRSUAPI_DS_REPLICA_SYNC_WRITEABLE |
-                DRSUAPI_DS_REPLICA_SYNC_ALL_SOURCES;
-        
-
-        req = dcerpc_drsuapi_DsReplicaSync_send(drsuapi->pipe, r, r);
-        composite_continue_rpc(c, req, dreplsrv_op_notify_replica_sync_recv, st);
-}
-
-/*
-  called when we have an established connection
- */
-static void dreplsrv_op_notify_connect_recv(struct composite_context *creq)
-{
-        struct dreplsrv_op_notify_state *st = talloc_get_type(creq->async.private_data,
-                                                              struct dreplsrv_op_notify_state);
-        struct composite_context *c = st->creq;
-
-        c->status = dreplsrv_out_drsuapi_recv(creq);
-        if (!composite_is_ok(c)) return;
-
-        dreplsrv_op_notify_replica_sync_send(st);
-}
-
-/*
-  start the ReplicaSync async call
- */
-static struct composite_context *dreplsrv_op_notify_send(struct dreplsrv_notify_operation *op)
-{
-        struct composite_context *c;
-        struct composite_context *creq;
-        struct dreplsrv_op_notify_state *st;
-
-        c = composite_create(op, op->service->task->event_ctx);
-        if (c == NULL) return NULL;
-
-        st = talloc_zero(c, struct dreplsrv_op_notify_state);
-        if (composite_nomem(st, c)) return c;
-
-        st->creq        = c;
-        st->op          = op;
-
-        creq = dreplsrv_out_drsuapi_send(op->source_dsa->conn);
-        composite_continue(c, creq, dreplsrv_op_notify_connect_recv, st);
-
-        return c;
-}
-
-static void dreplsrv_notify_del_repsTo(struct dreplsrv_notify_operation *op)
-{
-        uint32_t count;
-        struct repsFromToBlob *reps;
-        WERROR werr;
-        struct dreplsrv_service *s = op->service;
-        int i;
-
-        werr = dsdb_loadreps(s->samdb, op, op->source_dsa->partition->dn, "repsTo", &reps, &count);
-        if (!W_ERROR_IS_OK(werr)) {
-                DEBUG(0,(__location__ ": Failed to load repsTo for %s\n",
-                         ldb_dn_get_linearized(op->source_dsa->partition->dn)));
-                return;
-        }
-
-        for (i=0; i<count; i++) {
-                if (GUID_compare(&reps[i].ctr.ctr1.source_dsa_obj_guid, 
-                                 &op->source_dsa->repsFrom1->source_dsa_obj_guid) == 0) {
-                        memmove(&reps[i], &reps[i+1],
-                                sizeof(reps[i])*(count-(i+1)));
-                        count--;
-                }
-        }
-
-        werr = dsdb_savereps(s->samdb, op, op->source_dsa->partition->dn, "repsTo", reps, count);
-        if (!W_ERROR_IS_OK(werr)) {
-                DEBUG(0,(__location__ ": Failed to save repsTo for %s\n",
-                         ldb_dn_get_linearized(op->source_dsa->partition->dn)));
-                return;
-        }
+        r->in.req->req1.naming_context = &partition->nc;
+        r->in.req->req1.source_dsa_guid = state->op->service->ntds_guid;
+        r->in.req->req1.options =
+                DRSUAPI_DRS_ASYNC_OP |
+                DRSUAPI_DRS_UPDATE_NOTIFICATION |
+                DRSUAPI_DRS_WRIT_REP;
+
+        if (state->op->is_urgent) {
+                r->in.req->req1.options |= DRSUAPI_DRS_SYNC_URGENT;
+        }
+
+        state->ndr_struct_ptr = r;
+
+        if (DEBUGLVL(10)) {
+                NDR_PRINT_IN_DEBUG(drsuapi_DsReplicaSync, r);
+        }
+
+        subreq = dcerpc_drsuapi_DsReplicaSync_r_send(state,
+                                                     state->ev,
+                                                     drsuapi->drsuapi_handle,
+                                                     r);
+        if (tevent_req_nomem(subreq, req)) {
+                return;
+        }
+        tevent_req_set_callback(subreq, dreplsrv_op_notify_replica_sync_done, req);
+}
+
+static void dreplsrv_op_notify_replica_sync_done(struct tevent_req *subreq)
+{
+        struct tevent_req *req =
+                tevent_req_callback_data(subreq,
+                struct tevent_req);
+        struct dreplsrv_op_notify_state *state =
+                tevent_req_data(req,
+                struct dreplsrv_op_notify_state);
+        struct drsuapi_DsReplicaSync *r = talloc_get_type(state->ndr_struct_ptr,
+                                                          struct drsuapi_DsReplicaSync);
+        NTSTATUS status;
+
+        state->ndr_struct_ptr = NULL;
+
+        status = dcerpc_drsuapi_DsReplicaSync_r_recv(subreq, r);
+        TALLOC_FREE(subreq);
+        if (tevent_req_nterror(req, status)) {
+                return;
+        }
+
+        if (!W_ERROR_IS_OK(r->out.result)) {
+                status = werror_to_ntstatus(r->out.result);
+                tevent_req_nterror(req, status);
+                return;
+        }
+
+        tevent_req_done(req);
+}
+
+static NTSTATUS dreplsrv_op_notify_recv(struct tevent_req *req)
+{
+        return tevent_req_simple_recv_ntstatus(req);
 }
 
@@ -173 +178 @@
   called when a notify operation has completed
  */
-static void dreplsrv_notify_op_callback(struct dreplsrv_notify_operation *op)
-{
+static void dreplsrv_notify_op_callback(struct tevent_req *subreq)
+{
+        struct dreplsrv_notify_operation *op =
+                tevent_req_callback_data(subreq,
+                struct dreplsrv_notify_operation);
         NTSTATUS status;
         struct dreplsrv_service *s = op->service;
-
-        status = composite_wait(op->creq);
+        WERROR werr;
+
+        status = dreplsrv_op_notify_recv(subreq);
+        werr = ntstatus_to_werror(status);
+        TALLOC_FREE(subreq);
         if (!NT_STATUS_IS_OK(status)) {
-                DEBUG(0,("dreplsrv_notify: Failed to send DsReplicaSync to %s for %s - %s\n",
+                DEBUG(4,("dreplsrv_notify: Failed to send DsReplicaSync to %s for %s - %s : %s\n",
                          op->source_dsa->repsFrom1->other_info->dns_name,
                          ldb_dn_get_linearized(op->source_dsa->partition->dn),
-                         nt_errstr(status)));
+                         nt_errstr(status), win_errstr(werr)));
         } else {
                 DEBUG(2,("dreplsrv_notify: DsReplicaSync OK for %s\n",
                          op->source_dsa->repsFrom1->other_info->dns_name));
                 op->source_dsa->notify_uSN = op->uSN;
-                /* delete the repsTo for this replication partner in the
-                   partition, as we have successfully told him to sync */
-                dreplsrv_notify_del_repsTo(op);
-        }
-        talloc_free(op->creq);
+        }
+
+        drepl_reps_update(s, "repsTo", op->source_dsa->partition->dn,
+                          &op->source_dsa->repsFrom1->source_dsa_obj_guid,
+                          werr);
 
         talloc_free(op);
         s->ops.n_current = NULL;
-        dreplsrv_notify_run_ops(s);
-}
-
-
-static void dreplsrv_notify_op_callback_creq(struct composite_context *creq)
-{
-        struct dreplsrv_notify_operation *op = talloc_get_type(creq->async.private_data,
-                                                               struct dreplsrv_notify_operation);
-        dreplsrv_notify_op_callback(op);
+        dreplsrv_run_pending_ops(s);
 }
 
@@ -213 +216 @@
 {
         struct dreplsrv_notify_operation *op;
+        struct tevent_req *subreq;
 
         if (s->ops.n_current || s->ops.current) {
@@ -228 +232 @@
         DLIST_REMOVE(s->ops.notifies, op);
 
-        op->creq = dreplsrv_op_notify_send(op);
-        if (!op->creq) {
-                dreplsrv_notify_op_callback(op);
-                return;
-        }
-
-        op->creq->async.fn              = dreplsrv_notify_op_callback_creq;
-        op->creq->async.private_data    = op;
+        subreq = dreplsrv_op_notify_send(op, s->task->event_ctx, op);
+        if (!subreq) {
+                DEBUG(0,("dreplsrv_notify_run_ops: dreplsrv_op_notify_send[%s][%s] - no memory\n",
+                         op->source_dsa->repsFrom1->other_info->dns_name,
+                         ldb_dn_get_linearized(op->source_dsa->partition->dn)));
+                return;
+        }
+        tevent_req_set_callback(subreq, dreplsrv_notify_op_callback, op);
+        DEBUG(4,("started DsReplicaSync for %s to %s\n",
+                 ldb_dn_get_linearized(op->source_dsa->partition->dn),
+                 op->source_dsa->repsFrom1->other_info->dns_name));
 }
 
@@ -242 +249 @@
   find a source_dsa for a given guid
  */
-static struct dreplsrv_partition_source_dsa *dreplsrv_find_source_dsa(struct dreplsrv_partition *p,
+static struct dreplsrv_partition_source_dsa *dreplsrv_find_notify_dsa(struct dreplsrv_partition *p,
                                                                       struct GUID *guid)
 {
         struct dreplsrv_partition_source_dsa *s;
 
+        /* first check the sources list */
         for (s=p->sources; s; s=s->next) {
+                if (GUID_compare(&s->repsFrom1->source_dsa_obj_guid, guid) == 0) {
+                        return s;
+                }
+        }
+
+        /* then the notifies list */
+        for (s=p->notifies; s; s=s->next) {
                 if (GUID_compare(&s->repsFrom1->source_dsa_obj_guid, guid) == 0) {
                         return s;
@@ -263 +278 @@
                                             struct repsFromToBlob *reps,
                                             TALLOC_CTX *mem_ctx,
-                                            uint64_t uSN)
+                                            uint64_t uSN,
+                                            bool is_urgent,
+                                            uint32_t replica_flags)
 {
         struct dreplsrv_notify_operation *op;
         struct dreplsrv_partition_source_dsa *s;
 
-        s = dreplsrv_find_source_dsa(p, &reps->ctr.ctr1.source_dsa_obj_guid);
+        s = dreplsrv_find_notify_dsa(p, &reps->ctr.ctr1.source_dsa_obj_guid);
         if (s == NULL) {
                 DEBUG(0,(__location__ ": Unable to find source_dsa for %s\n",
@@ -275 +292 @@
         }
 
+        /* first try to find an existing notify operation */
+        for (op = service->ops.notifies; op; op = op->next) {
+                if (op->source_dsa != s) {
+                        continue;
+                }
+
+                if (op->is_urgent != is_urgent) {
+                        continue;
+                }
+
+                if (op->replica_flags != replica_flags) {
+                        continue;
+                }
+
+                if (op->uSN < uSN) {
+                        op->uSN = uSN;
+                }
+
+                /* reuse the notify operation, as it's not yet started */
+                return WERR_OK;
+        }
+
         op = talloc_zero(mem_ctx, struct dreplsrv_notify_operation);
         W_ERROR_HAVE_NO_MEMORY(op);
 
-        op->service     = service;
-        op->source_dsa  = s;
-        op->uSN         = uSN;
+        op->service       = service;
+        op->source_dsa    = s;
+        op->uSN           = uSN;
+        op->is_urgent     = is_urgent;
+        op->replica_flags = replica_flags;
+        op->schedule_time = time(NULL);
 
         DLIST_ADD_END(service->ops.notifies, op, struct dreplsrv_notify_operation *);
@@ -298 +340 @@
         struct repsFromToBlob *reps;
         WERROR werr;
-        uint64_t uSN;
-        int ret, i;
+        uint64_t uSNHighest;
+        uint64_t uSNUrgent;
+        uint32_t i;
+        int ret;
 
         werr = dsdb_loadreps(s->samdb, mem_ctx, p->dn, "repsTo", &reps, &count);
-        if (count == 0) {
-                werr = dsdb_loadreps(s->samdb, mem_ctx, p->dn, "repsFrom", &reps, &count);
-        }
         if (!W_ERROR_IS_OK(werr)) {
                 DEBUG(0,(__location__ ": Failed to load repsTo for %s\n",
@@ -311 +352 @@
         }
 
-        /* loads the partition uSNHighest */
-        ret = dsdb_load_partition_usn(s->samdb, p->dn, &uSN);
-        if (ret != LDB_SUCCESS || uSN == 0) {
+        /* loads the partition uSNHighest and uSNUrgent */
+        ret = dsdb_load_partition_usn(s->samdb, p->dn, &uSNHighest, &uSNUrgent);
+        if (ret != LDB_SUCCESS || uSNHighest == 0) {
                 /* nothing to do */
                 return WERR_OK;
@@ -321 +362 @@
         for (i=0; i<count; i++) {
                 struct dreplsrv_partition_source_dsa *sdsa;
-                sdsa = dreplsrv_find_source_dsa(p, &reps[i].ctr.ctr1.source_dsa_obj_guid);
+                uint32_t replica_flags;
+                sdsa = dreplsrv_find_notify_dsa(p, &reps[i].ctr.ctr1.source_dsa_obj_guid);
+                replica_flags = reps[i].ctr.ctr1.replica_flags;
                 if (sdsa == NULL) continue;
-                if (sdsa->notify_uSN < uSN) {
+                if (sdsa->notify_uSN < uSNHighest) {
                         /* we need to tell this partner to replicate
                            with us */
-                        werr = dreplsrv_schedule_notify_sync(s, p, &reps[i], mem_ctx, uSN);
+                        bool is_urgent = sdsa->notify_uSN < uSNUrgent;
+
+                        /* check if urgent replication is needed */
+                        werr = dreplsrv_schedule_notify_sync(s, p, &reps[i], mem_ctx,
+                                                             uSNHighest, is_urgent, replica_flags);
                         if (!W_ERROR_IS_OK(werr)) {
                                 DEBUG(0,(__location__ ": Failed to setup notify to %s for %s\n",
@@ -333 +380 @@
                                 return werr;
                         }
+                        DEBUG(4,("queued DsReplicaSync for %s to %s (urgent=%s) uSN=%llu:%llu\n",
+                                 ldb_dn_get_linearized(p->dn),
+                                 reps[i].ctr.ctr1.other_info->dns_name,
+                                 is_urgent?"true":"false",
+                                 (unsigned long long)sdsa->notify_uSN,
+                                 (unsigned long long)uSNHighest));
                 }
         }
@@ -405 +458 @@
 
         tmp_mem = talloc_new(service);
-        DEBUG(2,("dreplsrv_notify_schedule(%u) %sscheduled for: %s\n",
+        DEBUG(4,("dreplsrv_notify_schedule(%u) %sscheduled for: %s\n",
                 next_interval,
                 (service->notify.te?"re":""),
@@ -426 +479 @@
 
         dreplsrv_run_pending_ops(service);
-        dreplsrv_notify_run_ops(service);
-}
+}

vendor/current/source4/dsdb/repl/drepl_out_helpers.c

@@ -25 +25 @@
 #include "smbd/service.h"
 #include "lib/events/events.h"
-#include "lib/messaging/irpc.h"
 #include "dsdb/repl/drepl_service.h"
-#include "lib/ldb/include/ldb_errors.h"
+#include <ldb_errors.h>
 #include "../lib/util/dlinklist.h"
 #include "librpc/gen_ndr/ndr_misc.h"
@@ -35 +34 @@
 #include "auth/gensec/gensec.h"
 #include "param/param.h"
+#include "../lib/util/tevent_ntstatus.h"
+#include "libcli/security/security.h"
 
 struct dreplsrv_out_drsuapi_state {
-        struct composite_context *creq;
+        struct tevent_context *ev;
 
         struct dreplsrv_out_connection *conn;
@@ -47 +48 @@
 };
 
-static void dreplsrv_out_drsuapi_connect_recv(struct composite_context *creq);
-
-struct composite_context *dreplsrv_out_drsuapi_send(struct dreplsrv_out_connection *conn)
-{
-        struct composite_context *c;
+static void dreplsrv_out_drsuapi_connect_done(struct composite_context *creq);
+
+struct tevent_req *dreplsrv_out_drsuapi_send(TALLOC_CTX *mem_ctx,
+                                             struct tevent_context *ev,
+                                             struct dreplsrv_out_connection *conn)
+{
+        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;
-
-        c->private_data = st;
-
-        st->creq        = c;
-        st->conn        = conn;
-        st->drsuapi     = conn->drsuapi;
-
-        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);
+
+        req = tevent_req_create(mem_ctx, &state,
+                                struct dreplsrv_out_drsuapi_state);
+        if (req == NULL) {
+                return NULL;
+        }
+
+        state->ev       = ev;
+        state->conn     = conn;
+        state->drsuapi  = conn->drsuapi;
+
+        if (state->drsuapi && !state->drsuapi->pipe->conn->dead) {
+                tevent_req_done(req);
+                return tevent_req_post(req, ev);
+        }
+
+        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;
-
-        creq = dcerpc_pipe_connect_b_send(st, conn->binding, &ndr_table_drsuapi,
+        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(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);
-
-        return c;
-}
-
-static void dreplsrv_out_drsuapi_bind_send(struct dreplsrv_out_drsuapi_state *st);
-
-static void dreplsrv_out_drsuapi_connect_recv(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;
-
-        dreplsrv_out_drsuapi_bind_send(st);
-}
-
-static void dreplsrv_out_drsuapi_bind_recv(struct rpc_request *req);
-
-static void dreplsrv_out_drsuapi_bind_send(struct dreplsrv_out_drsuapi_state *st)
-{
-        struct composite_context *c = st->creq;
-        struct rpc_request *req;
-
-        st->bind_info_ctr.length        = 28;
-        st->bind_info_ctr.info.info28   = st->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;
-
-        req = dcerpc_drsuapi_DsBind_send(st->drsuapi->pipe, st, &st->bind_r);
-        composite_continue_rpc(c, req, dreplsrv_out_drsuapi_bind_recv, st);
-}
-
-static void dreplsrv_out_drsuapi_bind_recv(struct rpc_request *req)
-{
-        struct dreplsrv_out_drsuapi_state *st = talloc_get_type(req->async.private_data,
-                                                struct dreplsrv_out_drsuapi_state);
-        struct composite_context *c = st->creq;
-
-        c->status = dcerpc_ndr_request_recv(req);
-        if (!composite_is_ok(c)) return;
-
-        if (!W_ERROR_IS_OK(st->bind_r.out.result)) {
-                composite_error(c, werror_to_ntstatus(st->bind_r.out.result));
-                return;
-        }
-
-        ZERO_STRUCT(st->drsuapi->remote_info28);
-        if (st->bind_r.out.bind_info) {
-                switch (st->bind_r.out.bind_info->length) {
+                                          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 req;
+}
+
+static void dreplsrv_out_drsuapi_bind_done(struct tevent_req *subreq);
+
+static void dreplsrv_out_drsuapi_connect_done(struct composite_context *creq)
+{
+        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;
+
+        status = dcerpc_pipe_connect_b_recv(creq,
+                                            state->drsuapi,
+                                            &state->drsuapi->pipe);
+        if (tevent_req_nterror(req, status)) {
+                return;
+        }
+
+        state->drsuapi->drsuapi_handle = state->drsuapi->pipe->binding_handle;
+
+        status = gensec_session_key(state->drsuapi->pipe->conn->security_state.generic_state,
+                                    &state->drsuapi->gensec_skey);
+        if (tevent_req_nterror(req, status)) {
+                return;
+        }
+
+        state->bind_info_ctr.length             = 28;
+        state->bind_info_ctr.info.info28        = state->conn->service->bind_info28;
+
+        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;
+
+        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_done(struct tevent_req *subreq)
+{
+        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;
+
+        status = dcerpc_drsuapi_DsBind_r_recv(subreq, state);
+        TALLOC_FREE(subreq);
+        if (tevent_req_nterror(req, status)) {
+                return;
+        }
+
+        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(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);
-}
-
-NTSTATUS dreplsrv_out_drsuapi_recv(struct composite_context *c)
-{
+        tevent_req_done(req);
+}
+
+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);
-        }
-
-        talloc_free(c);
-        return status;
+
+        if (tevent_req_is_nterror(req, &status)) {
+                tevent_req_received(req);
+                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);
-
-struct composite_context *dreplsrv_op_pull_source_send(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;
-
-        creq = dreplsrv_out_drsuapi_send(op->source_dsa->conn);
-        composite_continue(c, creq, dreplsrv_op_pull_source_connect_recv, st);
-
-        return c;
-}
-
-static void dreplsrv_op_pull_source_get_changes_send(struct dreplsrv_op_pull_source_state *st);
-
-static void dreplsrv_op_pull_source_connect_recv(struct composite_context *creq)
-{
-        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;
-
-        c->status = dreplsrv_out_drsuapi_recv(creq);
-        if (!composite_is_ok(c)) return;
-
-        dreplsrv_op_pull_source_get_changes_send(st);
-}
-
-static void dreplsrv_op_pull_source_get_changes_recv(struct rpc_request *req);
-
-static void dreplsrv_op_pull_source_get_changes_send(struct dreplsrv_op_pull_source_state *st)
-{
-        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;
+static void dreplsrv_op_pull_source_connect_done(struct tevent_req *subreq);
+
+struct tevent_req *dreplsrv_op_pull_source_send(TALLOC_CTX *mem_ctx,
+                                                struct tevent_context *ev,
+                                                struct dreplsrv_out_operation *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;
+
+        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 req;
+}
+
+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 tevent_req *req = tevent_req_callback_data(subreq,
+                                 struct tevent_req);
+        NTSTATUS status;
+
+        status = dreplsrv_out_drsuapi_recv(subreq);
+        TALLOC_FREE(subreq);
+        if (tevent_req_nterror(req, status)) {
+                return;
+        }
+
+        dreplsrv_op_pull_source_get_changes_trigger(req);
+}
+
+static void dreplsrv_op_pull_source_get_changes_done(struct tevent_req *subreq);
+
+/*
+  get a partial attribute set for a replication call
+ */
+static NTSTATUS dreplsrv_get_rodc_partial_attribute_set(struct dreplsrv_service *service,
+                                                        TALLOC_CTX *mem_ctx,
+                                                        struct drsuapi_DsPartialAttributeSet **_pas,
+                                                        bool for_schema)
+{
+        struct drsuapi_DsPartialAttributeSet *pas;
+        struct dsdb_schema *schema;
+        uint32_t i;
+
+        pas = talloc_zero(mem_ctx, struct drsuapi_DsPartialAttributeSet);
+        NT_STATUS_HAVE_NO_MEMORY(pas);
+
+        schema = dsdb_get_schema(service->samdb, NULL);
+
+        pas->version = 1;
+        pas->attids = talloc_array(pas, enum drsuapi_DsAttributeId, schema->num_attributes);
+        NT_STATUS_HAVE_NO_MEMORY_AND_FREE(pas->attids, pas);
+
+        for (i=0; i<schema->num_attributes; i++) {
+                struct dsdb_attribute *a;
+                a = schema->attributes_by_attributeID_id[i];
+                if (a->systemFlags & (DS_FLAG_ATTR_NOT_REPLICATED | DS_FLAG_ATTR_IS_CONSTRUCTED)) {
+                        continue;
+                }
+                if (a->searchFlags & SEARCH_FLAG_RODC_ATTRIBUTE) {
+                        continue;
+                }
+                pas->attids[pas->num_attids] = dsdb_attribute_get_attid(a, for_schema);
+                pas->num_attids++;
+        }
+        *_pas = pas;
+        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,
+                                                      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;
-
-        r = talloc(st, struct drsuapi_DsGetNCChanges);
-        if (composite_nomem(r, c)) return;
-
-        r->out.level_out = talloc(r, int32_t);
-        if (composite_nomem(r->out.level_out, c)) return;
+        struct drsuapi_DsReplicaCursorCtrEx *uptodateness_vector;
+        struct tevent_req *subreq;
+        struct drsuapi_DsPartialAttributeSet *pas = NULL;
+        NTSTATUS status;
+        uint32_t replica_flags;
+
+        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.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), win_errstr(werr)));
+                }
+        }
+
+        if (partition->uptodatevector_ex.count == 0) {
+                uptodateness_vector = NULL;
+        } else {
+                uptodateness_vector = &partition->uptodatevector_ex;
+        }
+
+        replica_flags = rf1->replica_flags;
+
+        if (service->am_rodc) {
+                bool for_schema = false;
+                if (ldb_dn_compare_base(ldb_get_schema_basedn(service->samdb), partition->dn) == 0) {
+                        for_schema = true;
+                }
+
+                status = dreplsrv_get_rodc_partial_attribute_set(service, r, &pas, for_schema);
+                if (!NT_STATUS_IS_OK(status)) {
+                        DEBUG(0,(__location__ ": Failed to construct partial attribute set : %s\n", nt_errstr(status)));
+                        return;
+                }
+                if (state->op->extended_op == DRSUAPI_EXOP_REPL_SECRET) {
+                        replica_flags &= ~DRSUAPI_DRS_SPECIAL_SECRET_PROCESSING;
+                }
+        }
 
         r->in.bind_handle       = &drsuapi->bind_handle;
@@ -262 +397 @@
                 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.replica_flags           = rf1->replica_flags;
+                r->in.req->req8.uptodateness_vector     = uptodateness_vector;
+                r->in.req->req8.replica_flags           = 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.partial_attribute_set   = NULL;
+                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   = pas;
                 r->in.req->req8.partial_attribute_set_ex= NULL;
                 r->in.req->req8.mapping_ctr.num_mappings= 0;
@@ -278 +413 @@
                 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.replica_flags           = rf1->replica_flags;
+                r->in.req->req5.uptodateness_vector     = uptodateness_vector;
+                r->in.req->req5.replica_flags           = 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;
-        }
-
-        req = dcerpc_drsuapi_DsGetNCChanges_send(drsuapi->pipe, r, r);
-        composite_continue_rpc(c, req, dreplsrv_op_pull_source_get_changes_recv, st);
-}
-
-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_get_changes_recv(struct rpc_request *req)
-{
-        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,
+                r->in.req->req5.extended_op             = state->op->extended_op;
+                r->in.req->req5.fsmo_info               = state->op->fsmo_info;
+        }
+
+#if 0
+        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_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_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);
+        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;
+        enum drsuapi_DsExtendedError extended_ret;
+        state->ndr_struct_ptr = NULL;
+
+        status = dcerpc_drsuapi_DsGetNCChanges_r_recv(subreq, r);
+        TALLOC_FREE(subreq);
+        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;
         }
@@ -338 +492 @@
                 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;
         }
@@ -349 +505 @@
         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);
-}
-
-static void dreplsrv_update_refs_send(struct dreplsrv_op_pull_source_state *st);
-
-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)
-{
-        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;
+                extended_ret = ctr6->extended_ret;
+        }
+
+        if (ctr_level == 1) {
+                extended_ret = ctr1->extended_ret;
+        }
+
+        if (state->op->extended_op != DRSUAPI_EXOP_NONE) {
+                state->op->extended_ret = extended_ret;
+
+                if (extended_ret != DRSUAPI_EXOP_ERR_SUCCESS) {
+                        status = NT_STATUS_UNSUCCESSFUL;
+                        tevent_req_nterror(req, status);
+                        return;
+                }
+        }
+
+        dreplsrv_op_pull_source_apply_changes_trigger(req, r, ctr_level, ctr1, 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 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 dsdb_schema *schema;
+        struct dsdb_schema *working_schema = NULL;
         const struct drsuapi_DsReplicaOIDMapping_Ctr *mapping_ctr;
         uint32_t object_count;
@@ -376 +551 @@
         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) {
@@ -401 +578 @@
                 break;
         default:
-                composite_error(c, werror_to_ntstatus(WERR_BAD_NET_RESP));
-                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, 
-                                                         &st->op->source_dsa->notify_uSN);
+                nt_status = werror_to_ntstatus(WERR_BAD_NET_RESP);
+                tevent_req_nterror(req, nt_status);
+                return;
+        }
+
+        schema = dsdb_get_schema(service->samdb, NULL);
+        if (!schema) {
+                DEBUG(0,(__location__ ": Schema is not loaded yet!\n"));
+                tevent_req_nterror(req, NT_STATUS_INTERNAL_ERROR);
+                return;
+        }
+
+        /*
+         * Decide what working schema to use for object conversion.
+         * We won't need a working schema for empty replicas sent.
+         */
+        if (first_object && ldb_dn_compare(partition->dn, schema->base_dn) == 0) {
+                /* create working schema to convert objects with */
+                status = dsdb_repl_make_working_schema(service->samdb,
+                                                       schema,
+                                                       mapping_ctr,
+                                                       object_count,
+                                                       first_object,
+                                                       &drsuapi->gensec_skey,
+                                                       state, &working_schema);
+                if (!W_ERROR_IS_OK(status)) {
+                        DEBUG(0,("Failed to create working schema: %s\n",
+                                 win_errstr(status)));
+                        tevent_req_nterror(req, NT_STATUS_INTERNAL_ERROR);
+                        return;
+                }
+        }
+
+        status = dsdb_replicated_objects_convert(service->samdb,
+                                                 working_schema ? working_schema : schema,
+                                                 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)) {
-                DEBUG(0,("Failed to commit objects: %s\n", win_errstr(status)));
-                composite_error(c, werror_to_ntstatus(status));
-                return;
-        }
-
-        /* if it applied fine, we need to update the highwatermark */
-        *st->op->source_dsa->repsFrom1 = rf1;
-
+                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_replicated_objects_commit(service->samdb,
+                                                working_schema,
+                                                objects,
+                                                &state->op->source_dsa->notify_uSN);
+        talloc_free(objects);
+        if (!W_ERROR_IS_OK(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 (state->op->extended_op == DRSUAPI_EXOP_NONE) {
+                /* if it applied fine, we need to update the highwatermark */
+                *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;
+        }
+
+        if (state->op->extended_op != DRSUAPI_EXOP_NONE ||
+            state->op->service->am_rodc) {
+                /*
+                  we don't do the UpdateRefs for extended ops or if we
+                  are a RODC
+                 */
+                tevent_req_done(req);
                 return;
         }
@@ -440 +675 @@
            so we can use the already established DRSUAPI pipe
         */
-        dreplsrv_update_refs_send(st);
-}
-
-WERROR dreplsrv_op_pull_source_recv(struct composite_context *c)
-{
-        NTSTATUS status;
-
-        status = composite_wait(c);
-
-        talloc_free(c);
-        return ntstatus_to_werror(status);
-}
-
-/*
-  receive a UpdateRefs reply
- */
-static void dreplsrv_update_refs_recv(struct rpc_request *req)
-{
-        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 drsuapi_DsReplicaUpdateRefs);
-
-        c->status = dcerpc_ndr_request_recv(req);
-        if (!composite_is_ok(c)) {
-                DEBUG(0,("UpdateRefs failed with %s\n", 
-                         nt_errstr(c->status)));
-                return;
-        }
-
-        if (!W_ERROR_IS_OK(r->out.result)) {
-                DEBUG(0,("UpdateRefs failed with %s for %s %s\n", 
-                         win_errstr(r->out.result),
-                         r->in.req.req1.dest_dsa_dns_name,
-                         r->in.req.req1.naming_context->dn));
-                composite_error(c, werror_to_ntstatus(r->out.result));
-                return;
-        }
-
-        DEBUG(4,("UpdateRefs OK for %s %s\n", 
-                 r->in.req.req1.dest_dsa_dns_name,
-                 r->in.req.req1.naming_context->dn));
-
-        composite_done(c);
-}
+        dreplsrv_update_refs_trigger(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_send(struct dreplsrv_op_pull_source_state *st)
-{
-        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;
+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;
-
-        r = talloc(st, struct drsuapi_DsReplicaUpdateRefs);
-        if (composite_nomem(r, c)) return;
+        struct tevent_req *subreq;
+
+        r = talloc(state, struct drsuapi_DsReplicaUpdateRefs);
+        if (tevent_req_nomem(r, req)) {
+                return;
+        }
 
         ntds_guid_str = GUID_string(r, &service->ntds_guid);
-        if (composite_nomem(ntds_guid_str, c)) return;
-
-        /* lp_realm() is not really right here */
+        if (tevent_req_nomem(ntds_guid_str, req)) {
+                return;
+        }
+
         ntds_dns_name = talloc_asprintf(r, "%s._msdcs.%s",
                                         ntds_guid_str,
-                                        lp_realm(service->task->lp_ctx));
-        if (composite_nomem(ntds_dns_name, c)) return;
+                                        lpcfg_dnsdomain(service->task->lp_ctx));
+        if (tevent_req_nomem(ntds_dns_name, req)) {
+                return;
+        }
 
         r->in.bind_handle       = &drsuapi->bind_handle;
@@ -518 +717 @@
         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 (!lp_parm_bool(service->task->lp_ctx, NULL, "repl", "RODC", false)) {
-                r->in.req.req1.options |= DRSUAPI_DS_REPLICA_UPDATE_WRITEABLE;
-        }
-
-        req = dcerpc_drsuapi_DsReplicaUpdateRefs_send(drsuapi->pipe, r, r);
-        composite_continue_rpc(c, req, dreplsrv_update_refs_recv, st);
-}
+        r->in.req.req1.options            = DRSUAPI_DRS_ADD_REF | DRSUAPI_DRS_DEL_REF;
+        if (!service->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;
+        }
+
+        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 tevent_req *req)
+{
+        NTSTATUS status;
+
+        if (tevent_req_is_nterror(req, &status)) {
+                tevent_req_received(req);
+                return ntstatus_to_werror(status);
+        }
+
+        tevent_req_received(req);
+        return WERR_OK;
+}
+

vendor/current/source4/dsdb/repl/drepl_out_pull.c

@@ -25 +25 @@
 #include "smbd/service.h"
 #include "lib/events/events.h"
-#include "lib/messaging/irpc.h"
 #include "dsdb/repl/drepl_service.h"
-#include "lib/ldb/include/ldb_errors.h"
+#include <ldb_errors.h>
 #include "../lib/util/dlinklist.h"
 #include "librpc/gen_ndr/ndr_misc.h"
@@ -33 +32 @@
 #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)
+#include "libcli/security/security.h"
+
+/*
+  update repsFrom/repsTo error information
+ */
+void drepl_reps_update(struct dreplsrv_service *s, const char *reps_attr,
+                       struct ldb_dn *dn,
+                       struct GUID *source_dsa_obj_guid, WERROR status)
+{
+        struct repsFromToBlob *reps;
+        uint32_t count, i;
+        WERROR werr;
+        TALLOC_CTX *tmp_ctx = talloc_new(s);
+        time_t t;
+        NTTIME now;
+
+        t = time(NULL);
+        unix_to_nt_time(&now, t);
+
+        werr = dsdb_loadreps(s->samdb, tmp_ctx, dn, reps_attr, &reps, &count);
+        if (!W_ERROR_IS_OK(werr)) {
+                talloc_free(tmp_ctx);
+                return;
+        }
+
+        for (i=0; i<count; i++) {
+                if (GUID_compare(source_dsa_obj_guid,
+                                 &reps[i].ctr.ctr1.source_dsa_obj_guid) == 0) {
+                        break;
+                }
+        }
+
+        if (i == count) {
+                /* no record to update */
+                talloc_free(tmp_ctx);
+                return;
+        }
+
+        /* only update the status fields */
+        reps[i].ctr.ctr1.last_attempt = now;
+        reps[i].ctr.ctr1.result_last_attempt = status;
+        if (W_ERROR_IS_OK(status)) {
+                reps[i].ctr.ctr1.last_success = now;
+                reps[i].ctr.ctr1.consecutive_sync_failures = 0;
+        } else {
+                reps[i].ctr.ctr1.consecutive_sync_failures++;
+        }
+
+        werr = dsdb_savereps(s->samdb, tmp_ctx, dn, reps_attr, reps, count);
+        if (!W_ERROR_IS_OK(werr)) {
+                DEBUG(2,("drepl_reps_update: Failed to save %s for %s: %s\n",
+                         reps_attr, ldb_dn_get_linearized(dn), win_errstr(werr)));
+        }
+        talloc_free(tmp_ctx);
+}
+
+WERROR dreplsrv_schedule_partition_pull_source(struct dreplsrv_service *s,
+                                               struct dreplsrv_partition_source_dsa *source,
+                                               uint32_t options,
+                                               enum drsuapi_DsExtendedOperation extended_op,
+                                               uint64_t fsmo_info,
+                                               dreplsrv_extended_callback_t callback,
+                                               void *cb_data)
 {
         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->options     = options;
+        op->extended_op = extended_op;
+        op->fsmo_info   = fsmo_info;
+        op->callback    = callback;
+        op->cb_data     = cb_data;
+        op->schedule_time = time(NULL);
 
         DLIST_ADD_END(s->ops.pending, op, struct dreplsrv_out_operation *);
-        talloc_steal(s, op);
+
         return WERR_OK;
 }
@@ -60 +123 @@
 
         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,
+                                                                 0, DRSUAPI_EXOP_NONE, 0,
+                                                                 NULL, NULL);
                 W_ERROR_NOT_OK_RETURN(status);
         }
@@ -81 +146 @@
 
 
-/* force an immediate of the specified partition by GUID  */
-WERROR dreplsrv_schedule_partition_pull_by_guid(struct dreplsrv_service *s, TALLOC_CTX *mem_ctx,
-                                                struct GUID *guid)
-{
-        struct dreplsrv_partition *p;
-        
-        for (p = s->partitions; p; p = p->next) {
-                if (GUID_compare(&p->nc.guid, guid) == 0) {
-                        return dreplsrv_schedule_partition_pull(s, p, mem_ctx);
-                }
-        }
-
-        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;
-        NTTIME now;
-
-        t = time(NULL);
-        unix_to_nt_time(&now, t);
-
-        rf->result_last_attempt = dreplsrv_op_pull_source_recv(op->creq);
-        if (W_ERROR_IS_OK(rf->result_last_attempt)) {
-                rf->consecutive_sync_failures   = 0;
-                rf->last_success                = now;
-                DEBUG(3,("dreplsrv_op_pull_source(%s)\n",
-                        win_errstr(rf->result_last_attempt)));
-                goto done;
-        }
-
-        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));
-
-done:
+        WERROR werr;
+
+        werr = dreplsrv_op_pull_source_recv(subreq);
+        TALLOC_FREE(subreq);
+
+        DEBUG(4,("dreplsrv_op_pull_source(%s) for %s\n", win_errstr(werr),
+                 ldb_dn_get_linearized(op->source_dsa->partition->dn)));
+
+        if (op->extended_op == DRSUAPI_EXOP_NONE) {
+                drepl_reps_update(s, "repsFrom", op->source_dsa->partition->dn,
+                                  &rf->source_dsa_obj_guid, werr);
+        }
+
+        if (op->callback) {
+                op->callback(s, werr, op->extended_ret, op->cb_data);
+        }
         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)
+}
+
+void dreplsrv_run_pull_ops(struct dreplsrv_service *s)
 {
         struct dreplsrv_out_operation *op;
         time_t t;
         NTTIME now;
-
-        if (s->ops.current || s->ops.n_current) {
+        struct tevent_req *subreq;
+        WERROR werr;
+
+        if (s->ops.current) {
                 /* if there's still one running, we're done */
                 return;
@@ -161 +200 @@
         op->source_dsa->repsFrom1->last_attempt = now;
 
-        op->creq = dreplsrv_op_pull_source_send(op);
-        if (!op->creq) {
-                dreplsrv_pending_op_callback(op);
-                return;
-        }
-
-        op->creq->async.fn              = dreplsrv_pending_op_callback_creq;
-        op->creq->async.private_data    = op;
-}
+        /* check if inbound replication is enabled */
+        if (!(op->options & DRSUAPI_DRS_SYNC_FORCED)) {
+                uint32_t rep_options;
+                if (samdb_ntds_options(op->service->samdb, &rep_options) != LDB_SUCCESS) {
+                        werr = WERR_DS_DRA_INTERNAL_ERROR;
+                        goto failed;
+                }
+
+                if ((rep_options & DS_NTDSDSA_OPT_DISABLE_INBOUND_REPL)) {
+                        werr = WERR_DS_DRA_SINK_DISABLED;
+                        goto failed;
+                }
+        }
+
+        subreq = dreplsrv_op_pull_source_send(op, s->task->event_ctx, op);
+        if (!subreq) {
+                werr = WERR_NOMEM;
+                goto failed;
+        }
+
+        tevent_req_set_callback(subreq, dreplsrv_pending_op_callback, op);
+        return;
+
+failed:
+        if (op->extended_op == DRSUAPI_EXOP_NONE) {
+                drepl_reps_update(s, "repsFrom", op->source_dsa->partition->dn,
+                                  &op->source_dsa->repsFrom1->source_dsa_obj_guid, werr);
+        }
+        /* unblock queue processing */
+        s->ops.current = NULL;
+        /*
+         * let the callback do its job just like in any other failure situation
+         */
+        if (op->callback) {
+                op->callback(s, werr, op->extended_ret, op->cb_data);
+        }
+}

vendor/current/source4/dsdb/repl/drepl_partitions.c

@@ -25 +25 @@
 #include "smbd/service.h"
 #include "lib/events/events.h"
-#include "lib/messaging/irpc.h"
 #include "dsdb/repl/drepl_service.h"
-#include "lib/ldb/include/ldb_errors.h"
+#include <ldb_errors.h>
 #include "../lib/util/dlinklist.h"
 #include "librpc/gen_ndr/ndr_misc.h"
 #include "librpc/gen_ndr/ndr_drsuapi.h"
 #include "librpc/gen_ndr/ndr_drsblobs.h"
+#include "libcli/security/security.h"
 #include "param/param.h"
 
@@ -37 +37 @@
 {
         WERROR status;
-        struct ldb_dn *basedn;
-        struct ldb_result *r;
+        static const char *attrs[] = { "namingContexts", NULL };
+        unsigned int i;
+        int ret;
+        TALLOC_CTX *tmp_ctx;
+        struct ldb_result *res;
         struct ldb_message_element *el;
-        static const char *attrs[] = { "namingContexts", NULL };
-        uint32_t i;
-        int ret;
-
-        basedn = ldb_dn_new(s, s->samdb, NULL);
-        W_ERROR_HAVE_NO_MEMORY(basedn);
-
-        ret = ldb_search(s->samdb, s, &r, basedn, LDB_SCOPE_BASE, attrs,
-                         "(objectClass=*)");
-        talloc_free(basedn);
+
+        tmp_ctx = talloc_new(s);
+        W_ERROR_HAVE_NO_MEMORY(tmp_ctx);
+
+        ret = ldb_search(s->samdb, tmp_ctx, &res,
+                         ldb_dn_new(tmp_ctx, s->samdb, ""), LDB_SCOPE_BASE, attrs, NULL);
         if (ret != LDB_SUCCESS) {
-                return WERR_FOOBAR;
-        } else if (r->count != 1) {
-                talloc_free(r);
-                return WERR_FOOBAR;
-        }
-
-        el = ldb_msg_find_element(r->msgs[0], "namingContexts");
-        if (!el) {
-                return WERR_FOOBAR;
-        }
-
-        for (i=0; el && i < el->num_values; i++) {
-                const char *v = (const char *)el->values[i].data;
-                struct ldb_dn *pdn;
-                struct dreplsrv_partition *p;
-
-                pdn = ldb_dn_new(s, s->samdb, v);
-                if (!ldb_dn_validate(pdn)) {
-                        return WERR_FOOBAR;
-                }
-
-                p = talloc_zero(s, struct dreplsrv_partition);
-                W_ERROR_HAVE_NO_MEMORY(p);
-
-                p->dn = talloc_steal(p, pdn);
-
-                DLIST_ADD(s->partitions, p);
-
-                DEBUG(2, ("dreplsrv_partition[%s] loaded\n", v));
-        }
-
-        talloc_free(r);
+                DEBUG(1,("Searching for namingContexts in rootDSE failed: %s\n", ldb_errstring(s->samdb)));
+                talloc_free(tmp_ctx);
+                return WERR_DS_DRA_INTERNAL_ERROR;
+       }
+
+       el = ldb_msg_find_element(res->msgs[0], "namingContexts");
+       if (!el) {
+               DEBUG(1,("Finding namingContexts element in root_res failed: %s\n",
+                        ldb_errstring(s->samdb)));
+               talloc_free(tmp_ctx);
+               return WERR_DS_DRA_INTERNAL_ERROR;
+       }
+
+       for (i=0; i<el->num_values; i++) {
+               struct ldb_dn *pdn;
+               struct dreplsrv_partition *p;
+
+               pdn = ldb_dn_from_ldb_val(tmp_ctx, s->samdb, &el->values[i]);
+               if (pdn == NULL) {
+                       talloc_free(tmp_ctx);
+                       return WERR_DS_DRA_INTERNAL_ERROR;
+               }
+               if (!ldb_dn_validate(pdn)) {
+                       return WERR_DS_DRA_INTERNAL_ERROR;
+               }
+
+               p = talloc_zero(s, struct dreplsrv_partition);
+               W_ERROR_HAVE_NO_MEMORY(p);
+
+               p->dn = talloc_steal(p, pdn);
+
+               DLIST_ADD(s->partitions, p);
+
+               DEBUG(2, ("dreplsrv_partition[%s] loaded\n", ldb_dn_get_linearized(p->dn)));
+        }
+
+        talloc_free(tmp_ctx);
 
         status = dreplsrv_refresh_partitions(s);
@@ -90 +94 @@
 }
 
-static WERROR dreplsrv_out_connection_attach(struct dreplsrv_service *s,
-                                             const struct repsFromTo1 *rft,
-                                             struct dreplsrv_out_connection **_conn)
+/*
+  work out the principal to use for DRS replication connections
+ */
+NTSTATUS dreplsrv_get_target_principal(struct dreplsrv_service *s,
+                                       TALLOC_CTX *mem_ctx,
+                                       const struct repsFromTo1 *rft,
+                                       const char **target_principal)
+{
+        TALLOC_CTX *tmp_ctx;
+        struct ldb_result *res;
+        const char *attrs[] = { "dNSHostName", NULL };
+        int ret;
+        const char *hostname;
+        struct ldb_dn *dn;
+
+        *target_principal = NULL;
+
+        tmp_ctx = talloc_new(mem_ctx);
+
+        /* we need to find their hostname */
+        ret = dsdb_find_dn_by_guid(s->samdb, tmp_ctx, &rft->source_dsa_obj_guid, &dn);
+        if (ret != LDB_SUCCESS) {
+                talloc_free(tmp_ctx);
+                /* its OK for their NTDSDSA DN not to be in our database */
+                return NT_STATUS_OK;
+        }
+
+        /* strip off the NTDS Settings */
+        if (!ldb_dn_remove_child_components(dn, 1)) {
+                talloc_free(tmp_ctx);
+                return NT_STATUS_OK;
+        }
+
+        ret = dsdb_search_dn(s->samdb, tmp_ctx, &res, dn, attrs, 0);
+        if (ret != LDB_SUCCESS) {
+                talloc_free(tmp_ctx);
+                /* its OK for their account DN not to be in our database */
+                return NT_STATUS_OK;
+        }
+
+        hostname = ldb_msg_find_attr_as_string(res->msgs[0], "dNSHostName", NULL);
+        if (hostname == NULL) {
+                talloc_free(tmp_ctx);
+                /* its OK to not have a dnshostname */
+                return NT_STATUS_OK;
+        }
+
+        /* All DCs have the GC/hostname/realm name, but if some of the
+         * preconditions are not satisfied, then we will fall back to
+         * the
+         * E3514235-4B06-11D1-AB04-00C04FC2DCD2/${NTDSGUID}/${DNSDOMAIN}
+         * name.  This means that if a AD server has a dnsHostName set
+         * on it's record, it must also have GC/hostname/realm
+         * servicePrincipalName */
+
+        *target_principal = talloc_asprintf(mem_ctx, "GC/%s/%s",
+                                            hostname,
+                                            lpcfg_dnsdomain(s->task->lp_ctx));
+        talloc_free(tmp_ctx);
+        return NT_STATUS_OK;
+}
+
+
+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;
@@ -132 +199 @@
                 }
 
+                /* use the GC principal for DRS replication */
+                nt_status = dreplsrv_get_target_principal(s, conn->binding,
+                                                          rft, &conn->binding->target_principal);
+                if (!NT_STATUS_IS_OK(nt_status)) {
+                        return ntstatus_to_werror(nt_status);
+                }
+
                 DLIST_ADD_END(s->connections, conn, struct dreplsrv_out_connection *);
 
-                DEBUG(2,("dreplsrv_out_connection_attach(%s): create\n", conn->binding->host));
+                DEBUG(4,("dreplsrv_out_connection_attach(%s): create\n", conn->binding->host));
         } else {
-                DEBUG(2,("dreplsrv_out_connection_attach(%s): attach\n", conn->binding->host));
+                DEBUG(4,("dreplsrv_out_connection_attach(%s): attach\n", conn->binding->host));
         }
 
@@ -143 +217 @@
 }
 
+/*
+  find an existing source dsa in a list
+ */
+static struct dreplsrv_partition_source_dsa *dreplsrv_find_source_dsa(struct dreplsrv_partition_source_dsa *list,
+                                                                      struct GUID *guid)
+{
+        struct dreplsrv_partition_source_dsa *s;
+        for (s=list; s; s=s->next) {
+                if (GUID_compare(&s->repsFrom1->source_dsa_obj_guid, guid) == 0) {
+                        return s;
+                }
+        }
+        return NULL;    
+}
+
+
+
 static WERROR dreplsrv_partition_add_source_dsa(struct dreplsrv_service *s,
                                                 struct dreplsrv_partition *p,
+                                                struct dreplsrv_partition_source_dsa **listp,
+                                                struct dreplsrv_partition_source_dsa *check_list,
                                                 const struct ldb_val *val)
 {
@@ -155 +248 @@
 
         ndr_err = ndr_pull_struct_blob(val, source, 
-                                       lp_iconv_convenience(s->task->lp_ctx), &source->_repsFromBlob,
+                                       &source->_repsFromBlob,
                                        (ndr_pull_flags_fn_t)ndr_pull_repsFromToBlob);
         if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
@@ -174 +267 @@
         W_ERROR_NOT_OK_RETURN(status);
 
-        /* remove any existing source with the same GUID */
-        for (s2=p->sources; s2; s2=s2->next) {
+        if (check_list && 
+            dreplsrv_find_source_dsa(check_list, &source->repsFrom1->source_dsa_obj_guid)) {
+                /* its in the check list, don't add it again */
+                talloc_free(source);
+                return WERR_OK;
+        }
+
+        /* re-use an existing source if found */
+        for (s2=*listp; s2; s2=s2->next) {
                 if (GUID_compare(&s2->repsFrom1->source_dsa_obj_guid, 
                                  &source->repsFrom1->source_dsa_obj_guid) == 0) {
@@ -186 +286 @@
         }
 
-        DLIST_ADD_END(p->sources, source, struct dreplsrv_partition_source_dsa *);
+        DLIST_ADD_END(*listp, source, struct dreplsrv_partition_source_dsa *);
         return WERR_OK;
 }
+
+WERROR dreplsrv_partition_find_for_nc(struct dreplsrv_service *s,
+                                      const struct GUID *nc_guid,
+                                      const struct dom_sid *nc_sid,
+                                      const char *nc_dn_str,
+                                      struct dreplsrv_partition **_p)
+{
+        struct dreplsrv_partition *p;
+        bool valid_sid, valid_guid;
+        struct dom_sid null_sid;
+        ZERO_STRUCT(null_sid);
+
+        SMB_ASSERT(_p);
+
+        valid_sid  = nc_sid && !dom_sid_equal(&null_sid, nc_sid);
+        valid_guid = nc_guid && !GUID_all_zero(nc_guid);
+
+        if (!valid_sid && !valid_guid && !nc_dn_str) {
+                return WERR_DS_DRA_INVALID_PARAMETER;
+        }
+
+        for (p = s->partitions; p; p = p->next) {
+                if ((valid_guid && GUID_equal(&p->nc.guid, nc_guid))
+                    || strequal(p->nc.dn, nc_dn_str)
+                    || (valid_sid && dom_sid_equal(&p->nc.sid, nc_sid)))
+                {
+                        *_p = p;
+                        return WERR_OK;
+                }
+        }
+
+        return WERR_DS_DRA_BAD_NC;
+}
+
+WERROR dreplsrv_partition_source_dsa_by_guid(struct dreplsrv_partition *p,
+                                             const struct GUID *dsa_guid,
+                                             struct dreplsrv_partition_source_dsa **_dsa)
+{
+        struct dreplsrv_partition_source_dsa *dsa;
+
+        SMB_ASSERT(dsa_guid != NULL);
+        SMB_ASSERT(!GUID_all_zero(dsa_guid));
+        SMB_ASSERT(_dsa);
+
+        for (dsa = p->sources; dsa; dsa = dsa->next) {
+                if (GUID_equal(dsa_guid, &dsa->repsFrom1->source_dsa_obj_guid)) {
+                        *_dsa = dsa;
+                        return WERR_OK;
+                }
+        }
+
+        return WERR_DS_DRA_NO_REPLICA;
+}
+
+WERROR dreplsrv_partition_source_dsa_by_dns(const struct dreplsrv_partition *p,
+                                            const char *dsa_dns,
+                                            struct dreplsrv_partition_source_dsa **_dsa)
+{
+        struct dreplsrv_partition_source_dsa *dsa;
+
+        SMB_ASSERT(dsa_dns != NULL);
+        SMB_ASSERT(_dsa);
+
+        for (dsa = p->sources; dsa; dsa = dsa->next) {
+                if (strequal(dsa_dns, dsa->repsFrom1->other_info->dns_name)) {
+                        *_dsa = dsa;
+                        return WERR_OK;
+                }
+        }
+
+        return WERR_DS_DRA_NO_REPLICA;
+}
+
 
 static WERROR dreplsrv_refresh_partition(struct dreplsrv_service *s,
@@ -194 +367 @@
 {
         WERROR status;
-        const struct ldb_val *ouv_value;
-        struct replUpToDateVectorBlob ouv;
         struct dom_sid *nc_sid;
         struct ldb_message_element *orf_el = NULL;
         struct ldb_result *r;
-        uint32_t i;
+        unsigned int i;
         int ret;
         TALLOC_CTX *mem_ctx = talloc_new(p);
@@ -205 +376 @@
                 "objectSid",
                 "objectGUID",
-                "replUpToDateVector",
                 "repsFrom",
+                "repsTo",
                 NULL
         };
 
-        DEBUG(2, ("dreplsrv_refresh_partition(%s)\n",
+        DEBUG(4, ("dreplsrv_refresh_partition(%s)\n",
                 ldb_dn_get_linearized(p->dn)));
 
@@ -216 +387 @@
                          "(objectClass=*)");
         if (ret != LDB_SUCCESS) {
-                talloc_free(mem_ctx);
-                return WERR_FOOBAR;
-        } else if (r->count != 1) {
                 talloc_free(mem_ctx);
                 return WERR_FOOBAR;
@@ -234 +402 @@
         }
 
-        ouv_value = ldb_msg_find_ldb_val(r->msgs[0], "replUpToDateVector");
-        if (ouv_value) {
-                enum ndr_err_code ndr_err;
-                ndr_err = ndr_pull_struct_blob(ouv_value, mem_ctx, 
-                                               lp_iconv_convenience(s->task->lp_ctx), &ouv,
-                                               (ndr_pull_flags_fn_t)ndr_pull_replUpToDateVectorBlob);
-                if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
-                        NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
-                        talloc_free(mem_ctx);
-                        return ntstatus_to_werror(nt_status);
-                }
-                /* NDR_PRINT_DEBUG(replUpToDateVectorBlob, &ouv); */
-                if (ouv.version != 2) {
-                        talloc_free(mem_ctx);
-                        return WERR_DS_DRA_INTERNAL_ERROR;
-                }
-
-                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
-         */
-
+        talloc_free(p->uptodatevector.cursors);
+        talloc_free(p->uptodatevector_ex.cursors);
+        ZERO_STRUCT(p->uptodatevector);
+        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) {
+                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");
         if (orf_el) {
                 for (i=0; i < orf_el->num_values; i++) {
-                        status = dreplsrv_partition_add_source_dsa(s, p, &orf_el->values[i]);
+                        status = dreplsrv_partition_add_source_dsa(s, p, &p->sources, 
+                                                                   NULL, &orf_el->values[i]);
+                        W_ERROR_NOT_OK_RETURN(status);  
+                }
+        }
+
+        orf_el = ldb_msg_find_element(r->msgs[0], "repsTo");
+        if (orf_el) {
+                for (i=0; i < orf_el->num_values; i++) {
+                        status = dreplsrv_partition_add_source_dsa(s, p, &p->notifies, 
+                                                                   p->sources, &orf_el->values[i]);
                         W_ERROR_NOT_OK_RETURN(status);  
                 }

vendor/current/source4/dsdb/repl/drepl_periodic.c

@@ -25 +25 @@
 #include "auth/auth.h"
 #include "smbd/service.h"
-#include "lib/messaging/irpc.h"
 #include "dsdb/repl/drepl_service.h"
-#include "lib/ldb/include/ldb_errors.h"
+#include <ldb_errors.h>
 #include "../lib/util/dlinklist.h"
 #include "librpc/gen_ndr/ndr_misc.h"
@@ -82 +81 @@
 
         tmp_mem = talloc_new(service);
-        DEBUG(2,("dreplsrv_periodic_schedule(%u) %sscheduled for: %s\n",
+        DEBUG(4,("dreplsrv_periodic_schedule(%u) %sscheduled for: %s\n",
                 next_interval,
                 (service->periodic.te?"re":""),
@@ -98 +97 @@
         TALLOC_CTX *mem_ctx;
 
-        DEBUG(2,("dreplsrv_periodic_run(): schedule pull replication\n"));
+        DEBUG(4,("dreplsrv_periodic_run(): schedule pull replication\n"));
+
+        /*
+         * KCC or some administrative tool
+         * might have changed Topology graph
+         * i.e. repsFrom/repsTo
+         */
+        dreplsrv_refresh_partitions(service);
 
         mem_ctx = talloc_new(service);
@@ -104 +110 @@
         talloc_free(mem_ctx);
 
-        DEBUG(2,("dreplsrv_periodic_run(): run pending_ops memory=%u\n", 
+        DEBUG(4,("dreplsrv_periodic_run(): run pending_ops memory=%u\n", 
                  (unsigned)talloc_total_blocks(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);
 }
+
+/*
+  run the next pending op, either a notify or a pull
+ */
+void dreplsrv_run_pending_ops(struct dreplsrv_service *s)
+{
+        if (!s->ops.notifies && !s->ops.pending) {
+                return;
+        }
+        if (!s->ops.notifies ||
+            (s->ops.pending &&
+             s->ops.notifies->schedule_time > s->ops.pending->schedule_time)) {
+                dreplsrv_run_pull_ops(s);
+        } else {
+                dreplsrv_notify_run_ops(s);
+        }
+}

vendor/current/source4/dsdb/repl/drepl_service.c

@@ -1 @@
-/* 
+/*
    Unix SMB/CIFS mplementation.
    DSDB replication service
-   
+
    Copyright (C) Stefan Metzmacher 2007
-    
+   Copyright (C) Kamen Mazdrashki <kamenim@samba.org> 2010
+
    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/>.
-   
+
 */
 
@@ -25 +26 @@
 #include "smbd/service.h"
 #include "lib/events/events.h"
-#include "lib/messaging/irpc.h"
 #include "dsdb/repl/drepl_service.h"
-#include "lib/ldb/include/ldb_errors.h"
+#include <ldb_errors.h>
 #include "../lib/util/dlinklist.h"
 #include "librpc/gen_ndr/ndr_misc.h"
 #include "librpc/gen_ndr/ndr_drsuapi.h"
 #include "librpc/gen_ndr/ndr_drsblobs.h"
+#include "librpc/gen_ndr/ndr_irpc.h"
 #include "param/param.h"
 
+/**
+ * Call-back data for _drepl_replica_sync_done_cb()
+ */
+struct drepl_replica_sync_cb_data {
+        struct irpc_message *msg;
+        struct drsuapi_DsReplicaSync *r;
+
+        /* number of ops left to be completed */
+        int ops_count;
+
+        /* last failure error code */
+        WERROR werr_last_failure;
+};
+
+
 static WERROR dreplsrv_init_creds(struct dreplsrv_service *service)
 {
-        NTSTATUS status;
-
-        status = auth_system_session_info(service, service->task->lp_ctx, 
-                                          &service->system_session_info);
-        if (!NT_STATUS_IS_OK(status)) {
-                return ntstatus_to_werror(status);
+        service->system_session_info = system_session(service->task->lp_ctx);
+        if (service->system_session_info == NULL) {
+                return WERR_NOMEM;
         }
 
@@ -52 +65 @@
         struct drsuapi_DsBindInfo28 *bind_info28;
 
-        service->samdb = samdb_connect(service, service->task->event_ctx, lp_ctx, service->system_session_info);
+        service->samdb = samdb_connect(service, service->task->event_ctx, lp_ctx, service->system_session_info, 0);
         if (!service->samdb) {
                 return WERR_DS_UNAVAILABLE;
@@ -61 +74 @@
                 return WERR_DS_UNAVAILABLE;
         }
-
         service->ntds_guid = *ntds_guid;
+
+        if (samdb_rodc(service->samdb, &service->am_rodc) != LDB_SUCCESS) {
+                DEBUG(0,(__location__ ": Failed to determine RODC status\n"));
+                return WERR_DS_UNAVAILABLE;
+        }
 
         bind_info28                             = &service->bind_info28;
@@ -84 +101 @@
         bind_info28->supported_extensions       |= DRSUAPI_SUPPORTED_EXTENSION_ADD_SID_HISTORY;
         bind_info28->supported_extensions       |= DRSUAPI_SUPPORTED_EXTENSION_POST_BETA3;
-        bind_info28->supported_extensions       |= DRSUAPI_SUPPORTED_EXTENSION_00100000;
+        bind_info28->supported_extensions       |= DRSUAPI_SUPPORTED_EXTENSION_GETCHGREQ_V5;
         bind_info28->supported_extensions       |= DRSUAPI_SUPPORTED_EXTENSION_GET_MEMBERSHIPS2;
         bind_info28->supported_extensions       |= DRSUAPI_SUPPORTED_EXTENSION_GETCHGREQ_V6;
@@ -106 +123 @@
 }
 
+
+/**
+ * Callback for dreplsrv_out_operation operation completion.
+ *
+ * We just need to complete a waiting IRPC message here.
+ * In case pull operation has failed,
+ * caller of this callback will dump
+ * failure information.
+ *
+ * NOTE: cb_data is allocated in IRPC msg's context
+ * and will be freed during irpc_send_reply() call.
+ */
+static void _drepl_replica_sync_done_cb(struct dreplsrv_service *service,
+                                        WERROR werr,
+                                        enum drsuapi_DsExtendedError ext_err,
+                                        void *cb_data)
+{
+        struct drepl_replica_sync_cb_data *data = talloc_get_type(cb_data,
+                                                                  struct drepl_replica_sync_cb_data);
+        struct irpc_message *msg = data->msg;
+        struct drsuapi_DsReplicaSync *r = data->r;
+
+        /* store last bad result */
+        if (!W_ERROR_IS_OK(werr)) {
+                data->werr_last_failure = werr;
+        }
+
+        /* decrement pending ops count */
+        data->ops_count--;
+
+        if (data->ops_count == 0) {
+                /* Return result to client */
+                r->out.result = data->werr_last_failure;
+
+                /* complete IRPC message */
+                irpc_send_reply(msg, NT_STATUS_OK);
+        }
+}
+
+/**
+ * Helper to schedule a replication operation with a source DSA.
+ * If 'data' is valid pointer, then a callback
+ * for the operation is passed and 'data->msg' is
+ * marked as 'deferred' - defer_reply = true
+ */
+static WERROR _drepl_schedule_replication(struct dreplsrv_service *service,
+                                          struct dreplsrv_partition_source_dsa *dsa,
+                                          struct drsuapi_DsReplicaObjectIdentifier *nc,
+                                          uint32_t rep_options,
+                                          struct drepl_replica_sync_cb_data *data,
+                                          TALLOC_CTX *mem_ctx)
+{
+        WERROR werr;
+        dreplsrv_extended_callback_t fn_callback = NULL;
+
+        if (data) {
+                fn_callback = _drepl_replica_sync_done_cb;
+        }
+
+        /* schedule replication item */
+        werr = dreplsrv_schedule_partition_pull_source(service, dsa, rep_options,
+                                                       DRSUAPI_EXOP_NONE, 0,
+                                                       fn_callback, data);
+        if (!W_ERROR_IS_OK(werr)) {
+                DEBUG(0,("%s: failed setup of sync of partition (%s, %s, %s) - %s\n",
+                         __FUNCTION__,
+                         GUID_string(mem_ctx, &nc->guid),
+                         nc->dn,
+                         dsa->repsFrom1->other_info->dns_name,
+                         win_errstr(werr)));
+                return werr;
+        }
+        /* log we've scheduled a replication item */
+        DEBUG(3,("%s: forcing sync of partition (%s, %s, %s)\n",
+                 __FUNCTION__,
+                 GUID_string(mem_ctx, &nc->guid),
+                 nc->dn,
+                 dsa->repsFrom1->other_info->dns_name));
+
+        /* mark IRPC message as deferred if necessary */
+        if (data) {
+                data->ops_count++;
+                data->msg->defer_reply = true;
+        }
+
+        return WERR_OK;
+}
+
 /*
   DsReplicaSync messages from the DRSUAPI server are forwarded here
  */
-static NTSTATUS drepl_replica_sync(struct irpc_message *msg, 
+static NTSTATUS drepl_replica_sync(struct irpc_message *msg,
                                    struct drsuapi_DsReplicaSync *r)
 {
+        WERROR werr;
+        struct dreplsrv_partition *p;
+        struct drepl_replica_sync_cb_data *cb_data;
+        struct dreplsrv_partition_source_dsa *dsa;
+        struct drsuapi_DsReplicaSyncRequest1 *req1;
+        struct drsuapi_DsReplicaObjectIdentifier *nc;
         struct dreplsrv_service *service = talloc_get_type(msg->private_data,
                                                            struct dreplsrv_service);
-        struct GUID *guid = &r->in.req.req1.naming_context->guid;
-
-        r->out.result = dreplsrv_schedule_partition_pull_by_guid(service, msg, guid);
-        if (W_ERROR_IS_OK(r->out.result)) {
-                DEBUG(3,("drepl_replica_sync: forcing sync of partition %s\n",
-                         GUID_string(msg, guid)));
-                dreplsrv_run_pending_ops(service);
+
+#define REPLICA_SYNC_FAIL(_msg, _werr) do {\
+                if (!W_ERROR_IS_OK(_werr)) { \
+                        DEBUG(0,(__location__ ": Failure - %s. werr = %s\n", \
+                                 _msg, win_errstr(_werr))); \
+                        NDR_PRINT_IN_DEBUG(drsuapi_DsReplicaSync, r); \
+                } \
+                r->out.result = _werr; \
+                goto done;\
+        } while(0)
+
+
+        if (r->in.level != 1) {
+                REPLICA_SYNC_FAIL("Unsupported level",
+                                  WERR_DS_DRA_INVALID_PARAMETER);
+        }
+
+        req1 = &r->in.req->req1;
+        nc   = req1->naming_context;
+
+        /* Check input parameters */
+        if (!nc) {
+                REPLICA_SYNC_FAIL("Invalid Naming Context",
+                                  WERR_DS_DRA_INVALID_PARAMETER);
+        }
+
+        /* Find Naming context to be synchronized */
+        werr = dreplsrv_partition_find_for_nc(service,
+                                              &nc->guid, &nc->sid, nc->dn,
+                                              &p);
+        if (!W_ERROR_IS_OK(werr)) {
+                REPLICA_SYNC_FAIL("Failed to find requested Naming Context",
+                                  werr);
+        }
+
+        /* should we process it asynchronously? */
+        if (req1->options & DRSUAPI_DRS_ASYNC_OP) {
+                cb_data = NULL;
         } else {
-                DEBUG(3,("drepl_replica_sync: failed setup of sync of partition %s - %s\n",
-                         GUID_string(msg, guid), win_errstr(r->out.result)));
-        }
+                cb_data = talloc_zero(msg, struct drepl_replica_sync_cb_data);
+                if (!cb_data) {
+                        REPLICA_SYNC_FAIL("Not enought memory",
+                                          WERR_DS_DRA_INTERNAL_ERROR);
+                }
+
+                cb_data->msg = msg;
+                cb_data->r   = r;
+                cb_data->werr_last_failure = WERR_OK;
+        }
+
+        /* collect source DSAs to sync with */
+        if (req1->options & DRSUAPI_DRS_SYNC_ALL) {
+                for (dsa = p->sources; dsa; dsa = dsa->next) {
+                        /* schedule replication item */
+                        werr = _drepl_schedule_replication(service, dsa, nc,
+                                                           req1->options, cb_data, msg);
+                        if (!W_ERROR_IS_OK(werr)) {
+                                REPLICA_SYNC_FAIL("_drepl_schedule_replication() failed",
+                                                  werr);
+                        }
+                }
+        } else {
+                if (req1->options & DRSUAPI_DRS_SYNC_BYNAME) {
+                        /* client should pass at least valid string */
+                        if (!req1->source_dsa_dns) {
+                                REPLICA_SYNC_FAIL("'source_dsa_dns' is not valid",
+                                                  WERR_DS_DRA_INVALID_PARAMETER);
+                        }
+
+                        werr = dreplsrv_partition_source_dsa_by_dns(p,
+                                                                    req1->source_dsa_dns,
+                                                                    &dsa);
+                } else {
+                        /* client should pass at least some GUID */
+                        if (GUID_all_zero(&req1->source_dsa_guid)) {
+                                REPLICA_SYNC_FAIL("'source_dsa_guid' is not valid",
+                                                  WERR_DS_DRA_INVALID_PARAMETER);
+                        }
+
+                        werr = dreplsrv_partition_source_dsa_by_guid(p,
+                                                                     &req1->source_dsa_guid,
+                                                                     &dsa);
+                }
+                if (!W_ERROR_IS_OK(werr)) {
+                        REPLICA_SYNC_FAIL("Failed to locate source DSA for given NC",
+                                          werr);
+                }
+
+                /* schedule replication item */
+                werr = _drepl_schedule_replication(service, dsa, nc,
+                                                   req1->options, cb_data, msg);
+                if (!W_ERROR_IS_OK(werr)) {
+                        REPLICA_SYNC_FAIL("_drepl_schedule_replication() failed",
+                                          werr);
+                }
+        }
+
+        /* if we got here, everything is OK */
+        r->out.result = WERR_OK;
+
+        /*
+         * schedule replication event to force
+         * replication as soon as possible
+         */
+        dreplsrv_periodic_schedule(service, 0);
+
+done:
         return NT_STATUS_OK;
 }
+
+/**
+ * Called when drplsrv should refresh its state.
+ * For example, when KCC change topology, dreplsrv
+ * should update its cache
+ *
+ * @param partition_dn If not empty/NULL, partition to update
+ */
+static NTSTATUS dreplsrv_refresh(struct irpc_message *msg,
+                                 struct dreplsrv_refresh *r)
+{
+        struct dreplsrv_service *s = talloc_get_type(msg->private_data,
+                                                     struct dreplsrv_service);
+
+        r->out.result = dreplsrv_refresh_partitions(s);
+
+        return NT_STATUS_OK;
+}
+
+static NTSTATUS drepl_take_FSMO_role(struct irpc_message *msg,
+                                     struct drepl_takeFSMORole *r)
+{
+        struct dreplsrv_service *service = talloc_get_type(msg->private_data,
+                                                           struct dreplsrv_service);
+        r->out.result = dreplsrv_fsmo_role_check(service, r->in.role);
+        return NT_STATUS_OK;
+}
+
+/**
+ * Called when the auth code wants us to try and replicate
+ * a users secrets
+ */
+static NTSTATUS drepl_trigger_repl_secret(struct irpc_message *msg,
+                                          struct drepl_trigger_repl_secret *r)
+{
+        struct dreplsrv_service *service = talloc_get_type(msg->private_data,
+                                                           struct dreplsrv_service);
+
+
+        drepl_repl_secret(service, r->in.user_dn);
+
+        /* we are not going to be sending a reply to this request */
+        msg->no_reply = true;
+
+        return NT_STATUS_OK;
+}
+
+
+/*
+  DsReplicaAdd messages from the DRSUAPI server are forwarded here
+ */
+static NTSTATUS dreplsrv_replica_add(struct irpc_message *msg,
+                                  struct drsuapi_DsReplicaAdd *r)
+{
+        struct dreplsrv_service *service = talloc_get_type(msg->private_data,
+                                                           struct dreplsrv_service);
+        return drepl_replica_add(service, r);
+}
+
+/*
+  DsReplicaDel messages from the DRSUAPI server are forwarded here
+ */
+static NTSTATUS dreplsrv_replica_del(struct irpc_message *msg,
+                                  struct drsuapi_DsReplicaDel *r)
+{
+        struct dreplsrv_service *service = talloc_get_type(msg->private_data,
+                                                           struct dreplsrv_service);
+        return drepl_replica_del(service, r);
+}
+
+/*
+  DsReplicaMod messages from the DRSUAPI server are forwarded here
+ */
+static NTSTATUS dreplsrv_replica_mod(struct irpc_message *msg,
+                                  struct drsuapi_DsReplicaMod *r)
+{
+        struct dreplsrv_service *service = talloc_get_type(msg->private_data,
+                                                           struct dreplsrv_service);
+        return drepl_replica_mod(service, r);
+}
+
 
 /*
@@ -137 +425 @@
         uint32_t periodic_startup_interval;
 
-        switch (lp_server_role(task->lp_ctx)) {
+        switch (lpcfg_server_role(task->lp_ctx)) {
         case ROLE_STANDALONE:
-                task_server_terminate(task, "dreplsrv: no DSDB replication required in standalone configuration", 
+                task_server_terminate(task, "dreplsrv: no DSDB replication required in standalone configuration",
                                       false);
                 return;
         case ROLE_DOMAIN_MEMBER:
-                task_server_terminate(task, "dreplsrv: no DSDB replication required in domain member configuration", 
+                task_server_terminate(task, "dreplsrv: no DSDB replication required in domain member configuration",
                                       false);
                 return;
@@ -186 +474 @@
         }
 
-        periodic_startup_interval       = lp_parm_int(task->lp_ctx, NULL, "dreplsrv", "periodic_startup_interval", 15); /* in seconds */
-        service->periodic.interval      = lp_parm_int(task->lp_ctx, NULL, "dreplsrv", "periodic_interval", 300); /* in seconds */
+        periodic_startup_interval       = lpcfg_parm_int(task->lp_ctx, NULL, "dreplsrv", "periodic_startup_interval", 15); /* in seconds */
+        service->periodic.interval      = lpcfg_parm_int(task->lp_ctx, NULL, "dreplsrv", "periodic_interval", 300); /* in seconds */
 
         status = dreplsrv_periodic_schedule(service, periodic_startup_interval);
@@ -197 +485 @@
         }
 
-        service->notify.interval = lp_parm_int(task->lp_ctx, NULL, "dreplsrv", 
-                                               "notify_interval", 5); /* in seconds */
-        status = dreplsrv_notify_schedule(service, service->notify.interval);
-        if (!W_ERROR_IS_OK(status)) {
-                task_server_terminate(task, talloc_asprintf(task,
-                                      "dreplsrv: Failed to setup notify schedule: %s\n",
-                                                            win_errstr(status)), true);
-                return;
+        /* if we are a RODC then we do not send DSReplicaSync*/
+        if (!service->am_rodc) {
+                service->notify.interval = lpcfg_parm_int(task->lp_ctx, NULL, "dreplsrv",
+                                                           "notify_interval", 5); /* in seconds */
+                status = dreplsrv_notify_schedule(service, service->notify.interval);
+                if (!W_ERROR_IS_OK(status)) {
+                        task_server_terminate(task, talloc_asprintf(task,
+                                                  "dreplsrv: Failed to setup notify schedule: %s\n",
+                                                                        win_errstr(status)), true);
+                        return;
+                }
         }
 
         irpc_add_name(task->msg_ctx, "dreplsrv");
 
+        IRPC_REGISTER(task->msg_ctx, irpc, DREPLSRV_REFRESH, dreplsrv_refresh, service);
         IRPC_REGISTER(task->msg_ctx, drsuapi, DRSUAPI_DSREPLICASYNC, drepl_replica_sync, service);
+        IRPC_REGISTER(task->msg_ctx, drsuapi, DRSUAPI_DSREPLICAADD, dreplsrv_replica_add, service);
+        IRPC_REGISTER(task->msg_ctx, drsuapi, DRSUAPI_DSREPLICADEL, dreplsrv_replica_del, service);
+        IRPC_REGISTER(task->msg_ctx, drsuapi, DRSUAPI_DSREPLICAMOD, dreplsrv_replica_mod, service);
+        IRPC_REGISTER(task->msg_ctx, irpc, DREPL_TAKEFSMOROLE, drepl_take_FSMO_role, service);
+        IRPC_REGISTER(task->msg_ctx, irpc, DREPL_TRIGGER_REPL_SECRET, drepl_trigger_repl_secret, service);
+        messaging_register(task->msg_ctx, service, MSG_DREPL_ALLOCATE_RID, dreplsrv_allocate_rid);
 }
 

vendor/current/source4/dsdb/repl/drepl_service.h

@@ -34 +34 @@
          */
         struct dcerpc_pipe *pipe;
+        struct dcerpc_binding_handle *drsuapi_handle;
 
         DATA_BLOB gensec_skey;
@@ -52 +53 @@
         /* the out going connection to the source dsa */
         struct dreplsrv_drsuapi_connection *drsuapi;
+
+        /* used to force the GC principal name */
+        const char *principal_name;
 };
 
@@ -99 +103 @@
          */
         struct dreplsrv_partition_source_dsa *sources;
-};
+
+        /*
+         * a linked list of all source dsa's we will notify,
+         * that are not also in sources
+         */
+        struct dreplsrv_partition_source_dsa *notifies;
+
+        bool incoming_only;
+};
+
+typedef void (*dreplsrv_extended_callback_t)(struct dreplsrv_service *,
+                                             WERROR,
+                                             enum drsuapi_DsExtendedError,
+                                             void *cb_data);
 
 struct dreplsrv_out_operation {
         struct dreplsrv_out_operation *prev, *next;
+        time_t schedule_time;
 
         struct dreplsrv_service *service;
@@ -108 +126 @@
         struct dreplsrv_partition_source_dsa *source_dsa;
 
-        struct composite_context *creq;
+        /* replication options - currently used by DsReplicaSync */
+        uint32_t options;
+        enum drsuapi_DsExtendedOperation extended_op;
+        uint64_t fsmo_info;
+        enum drsuapi_DsExtendedError extended_ret;
+        dreplsrv_extended_callback_t callback;
+        void *cb_data;
 };
 
 struct dreplsrv_notify_operation {
         struct dreplsrv_notify_operation *prev, *next;
+        time_t schedule_time;
 
         struct dreplsrv_service *service;
@@ -118 +143 @@
 
         struct dreplsrv_partition_source_dsa *source_dsa;
-
-        struct composite_context *creq;
+        bool is_urgent;
+        uint32_t replica_flags;
 };
 
@@ -205 +230 @@
                 struct dreplsrv_notify_operation *n_current;
         } ops;
-};
-
+
+        bool rid_alloc_in_progress;
+
+        bool am_rodc;
+};
+
+#include "lib/messaging/irpc.h"
 #include "dsdb/repl/drepl_out_helpers.h"
 #include "dsdb/repl/drepl_service_proto.h"

vendor/current/source4/dsdb/repl/replicated_objects.c

@@ -22 +22 @@
 #include "includes.h"
 #include "dsdb/samdb/samdb.h"
-#include "lib/ldb/include/ldb_errors.h"
+#include <ldb_errors.h>
 #include "../lib/util/dlinklist.h"
 #include "librpc/gen_ndr/ndr_misc.h"
@@ -32 +32 @@
 #include "param/param.h"
 
-static WERROR dsdb_convert_object_ex(struct ldb_context *ldb,
-                                     const struct dsdb_schema *schema,
-                                     const struct drsuapi_DsReplicaObjectListItemEx *in,
+/**
+ * Multi-pass working schema creation
+ * Function will:
+ *  - shallow copy initial schema supplied
+ *  - create a working schema in multiple passes
+ *    until all objects are resolved
+ * Working schema is a schema with Attributes, Classes
+ * and indexes, but w/o subClassOf, possibleSupperiors etc.
+ * It is to be used just us cache for converting attribute values.
+ */
+WERROR dsdb_repl_make_working_schema(struct ldb_context *ldb,
+                                     const struct dsdb_schema *initial_schema,
+                                     const struct drsuapi_DsReplicaOIDMapping_Ctr *mapping_ctr,
+                                     uint32_t object_count,
+                                     const struct drsuapi_DsReplicaObjectListItemEx *first_object,
                                      const DATA_BLOB *gensec_skey,
                                      TALLOC_CTX *mem_ctx,
-                                     struct dsdb_extended_replicated_object *out)
+                                     struct dsdb_schema **_schema_out)
+{
+        struct schema_list {
+                struct schema_list *next, *prev;
+                const struct drsuapi_DsReplicaObjectListItemEx *obj;
+        };
+
+        WERROR werr;
+        struct dsdb_schema_prefixmap *pfm_remote;
+        struct schema_list *schema_list = NULL, *schema_list_item, *schema_list_next_item;
+        struct dsdb_schema *working_schema;
+        const struct drsuapi_DsReplicaObjectListItemEx *cur;
+        int ret, pass_no;
+        uint32_t ignore_attids[] = {
+                        DRSUAPI_ATTID_auxiliaryClass,
+                        DRSUAPI_ATTID_mayContain,
+                        DRSUAPI_ATTID_mustContain,
+                        DRSUAPI_ATTID_possSuperiors,
+                        DRSUAPI_ATTID_systemPossSuperiors,
+                        DRSUAPI_ATTID_INVALID
+        };
+
+        /* make a copy of the iniatial_scheam so we don't mess with it */
+        working_schema = dsdb_schema_copy_shallow(mem_ctx, ldb, initial_schema);
+        if (!working_schema) {
+                DEBUG(0,(__location__ ": schema copy failed!\n"));
+                return WERR_NOMEM;
+        }
+
+        /* we are going to need remote prefixMap for decoding */
+        werr = dsdb_schema_pfm_from_drsuapi_pfm(mapping_ctr, true,
+                                                mem_ctx, &pfm_remote, NULL);
+        if (!W_ERROR_IS_OK(werr)) {
+                DEBUG(0,(__location__ ": Failed to decode remote prefixMap: %s",
+                         win_errstr(werr)));
+                return werr;
+        }
+
+        /* create a list of objects yet to be converted */
+        for (cur = first_object; cur; cur = cur->next_object) {
+                schema_list_item = talloc(mem_ctx, struct schema_list);
+                schema_list_item->obj = cur;
+                DLIST_ADD_END(schema_list, schema_list_item, struct schema_list);
+        }
+
+        /* resolve objects until all are resolved and in local schema */
+        pass_no = 1;
+
+        while (schema_list) {
+                uint32_t converted_obj_count = 0;
+                uint32_t failed_obj_count = 0;
+                TALLOC_CTX *tmp_ctx = talloc_new(mem_ctx);
+                W_ERROR_HAVE_NO_MEMORY(tmp_ctx);
+
+                for (schema_list_item = schema_list; schema_list_item; schema_list_item=schema_list_next_item) {
+                        struct dsdb_extended_replicated_object object;
+
+                        cur = schema_list_item->obj;
+
+                        /* Save the next item, now we have saved out
+                         * the current one, so we can DLIST_REMOVE it
+                         * safely */
+                        schema_list_next_item = schema_list_item->next;
+
+                        /*
+                         * Convert the objects into LDB messages using the
+                         * schema we have so far. It's ok if we fail to convert
+                         * an object. We should convert more objects on next pass.
+                         */
+                        werr = dsdb_convert_object_ex(ldb, working_schema, pfm_remote,
+                                                      cur, gensec_skey,
+                                                      ignore_attids,
+                                                      tmp_ctx, &object);
+                        if (!W_ERROR_IS_OK(werr)) {
+                                DEBUG(1,("Warning: Failed to convert schema object %s into ldb msg\n",
+                                         cur->object.identifier->dn));
+
+                                failed_obj_count++;
+                        } else {
+                                /*
+                                 * Convert the schema from ldb_message format
+                                 * (OIDs as OID strings) into schema, using
+                                 * the remote prefixMap
+                                 */
+                                werr = dsdb_schema_set_el_from_ldb_msg(ldb,
+                                                                       working_schema,
+                                                                       object.msg);
+                                if (!W_ERROR_IS_OK(werr)) {
+                                        DEBUG(1,("Warning: failed to convert object %s into a schema element: %s\n",
+                                                 ldb_dn_get_linearized(object.msg->dn),
+                                                 win_errstr(werr)));
+                                        failed_obj_count++;
+                                } else {
+                                        DLIST_REMOVE(schema_list, schema_list_item);
+                                        talloc_free(schema_list_item);
+                                        converted_obj_count++;
+                                }
+                        }
+                }
+                talloc_free(tmp_ctx);
+
+                DEBUG(4,("Schema load pass %d: %d/%d of %d objects left to be converted.\n",
+                         pass_no, failed_obj_count, converted_obj_count, object_count));
+                pass_no++;
+
+                /* check if we converted any objects in this pass */
+                if (converted_obj_count == 0) {
+                        DEBUG(0,("Can't continue Schema load: didn't manage to convert any objects: all %d remaining of %d objects failed to convert\n", failed_obj_count, object_count));
+                        return WERR_INTERNAL_ERROR;
+                }
+
+                /* rebuild indexes */
+                ret = dsdb_setup_sorted_accessors(ldb, working_schema);
+                if (LDB_SUCCESS != ret) {
+                        DEBUG(0,("Failed to create schema-cache indexes!\n"));
+                        return WERR_INTERNAL_ERROR;
+                }
+        };
+
+        *_schema_out = working_schema;
+
+        return WERR_OK;
+}
+
+static bool dsdb_attid_in_list(const uint32_t attid_list[], uint32_t attid)
+{
+        const uint32_t *cur;
+        if (!attid_list) {
+                return false;
+        }
+        for (cur = attid_list; *cur != DRSUAPI_ATTID_INVALID; cur++) {
+                if (*cur == attid) {
+                        return true;
+                }
+        }
+        return false;
+}
+
+WERROR dsdb_convert_object_ex(struct ldb_context *ldb,
+                              const struct dsdb_schema *schema,
+                              const struct dsdb_schema_prefixmap *pfm_remote,
+                              const struct drsuapi_DsReplicaObjectListItemEx *in,
+                              const DATA_BLOB *gensec_skey,
+                              const uint32_t *ignore_attids,
+                              TALLOC_CTX *mem_ctx,
+                              struct dsdb_extended_replicated_object *out)
 {
         NTSTATUS nt_status;
-        enum ndr_err_code ndr_err;
         WERROR status;
         uint32_t i;
@@ -58 +214 @@
         struct dom_sid *sid = NULL;
         uint32_t rid = 0;
+        uint32_t attr_count;
         int ret;
 
@@ -112 +269 @@
         W_ERROR_HAVE_NO_MEMORY(md->ctr.ctr1.array);
 
-        for (i=0; i < in->meta_data_ctr->count; i++) {
+        for (i=0, attr_count=0; i < in->meta_data_ctr->count; i++, attr_count++) {
                 struct drsuapi_DsReplicaAttribute *a;
                 struct drsuapi_DsReplicaMetaData *d;
                 struct replPropertyMetaData1 *m;
                 struct ldb_message_element *e;
-                int j;
+                uint32_t j;
 
                 a = &in->object.attribute_ctr.attributes[i];
                 d = &in->meta_data_ctr->meta_data[i];
-                m = &md->ctr.ctr1.array[i];
-                e = &msg->elements[i];
+                m = &md->ctr.ctr1.array[attr_count];
+                e = &msg->elements[attr_count];
+
+                if (dsdb_attid_in_list(ignore_attids, a->attid)) {
+                        attr_count--;
+                        continue;
+                }
 
                 for (j=0; j<a->value_ctr.num_values; j++) {
@@ -129 +291 @@
                 }
 
-                status = dsdb_attribute_drsuapi_to_ldb(ldb, schema, a, msg->elements, e);
+                status = dsdb_attribute_drsuapi_to_ldb(ldb, schema, pfm_remote,
+                                                       a, msg->elements, e);
                 W_ERROR_NOT_OK_RETURN(status);
 
@@ -143 +306 @@
                 }
 
-                if (a->attid == DRSUAPI_ATTRIBUTE_name) {
+                if (a->attid == DRSUAPI_ATTID_name) {
                         name_a = a;
                         name_d = d;
-                        rdn_m = &md->ctr.ctr1.array[md->ctr.ctr1.count];
-                }
+                }
+        }
+
+        msg->num_elements = attr_count;
+        md->ctr.ctr1.count = attr_count;
+        if (name_a) {
+                rdn_m = &md->ctr.ctr1.array[md->ctr.ctr1.count];
         }
 
@@ -186 +354 @@
         W_ERROR_HAVE_NO_MEMORY(whenChanged_s);
 
-        ndr_err = ndr_push_struct_blob(&guid_value, msg, 
-                                       lp_iconv_convenience(ldb_get_opaque(ldb, "loadparm")),
-                                       &in->object.identifier->guid,
-                                         (ndr_push_flags_fn_t)ndr_push_GUID);
-        if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
-                nt_status = ndr_map_error2ntstatus(ndr_err);
+        nt_status = GUID_to_ndr_blob(&in->object.identifier->guid, msg, &guid_value);
+        if (!NT_STATUS_IS_OK(nt_status)) {
                 return ntstatus_to_werror(nt_status);
         }
@@ -202 +366 @@
 }
 
-WERROR dsdb_extended_replicated_objects_commit(struct ldb_context *ldb,
-                                               const char *partition_dn,
-                                               const struct drsuapi_DsReplicaOIDMapping_Ctr *mapping_ctr,
-                                               uint32_t object_count,
-                                               const struct drsuapi_DsReplicaObjectListItemEx *first_object,
-                                               uint32_t linked_attributes_count,
-                                               const struct drsuapi_DsReplicaLinkedAttribute *linked_attributes,
-                                               const struct repsFromTo1 *source_dsa,
-                                               const struct drsuapi_DsReplicaCursor2CtrEx *uptodateness_vector,
-                                               const DATA_BLOB *gensec_skey,
-                                               TALLOC_CTX *mem_ctx,
-                                               struct dsdb_extended_replicated_objects **_out,
-                                               uint64_t *notify_uSN)
+WERROR dsdb_replicated_objects_convert(struct ldb_context *ldb,
+                                       const struct dsdb_schema *schema,
+                                       const char *partition_dn_str,
+                                       const struct drsuapi_DsReplicaOIDMapping_Ctr *mapping_ctr,
+                                       uint32_t object_count,
+                                       const struct drsuapi_DsReplicaObjectListItemEx *first_object,
+                                       uint32_t linked_attributes_count,
+                                       const struct drsuapi_DsReplicaLinkedAttribute *linked_attributes,
+                                       const struct repsFromTo1 *source_dsa,
+                                       const struct drsuapi_DsReplicaCursor2CtrEx *uptodateness_vector,
+                                       const DATA_BLOB *gensec_skey,
+                                       TALLOC_CTX *mem_ctx,
+                                       struct dsdb_extended_replicated_objects **objects)
 {
         WERROR status;
-        const struct dsdb_schema *schema;
+        struct ldb_dn *partition_dn;
+        struct dsdb_schema_prefixmap *pfm_remote;
         struct dsdb_extended_replicated_objects *out;
-        struct ldb_result *ext_res;
         const struct drsuapi_DsReplicaObjectListItemEx *cur;
         uint32_t i;
-        int ret;
-        uint64_t seq_num1, seq_num2;
-
-        schema = dsdb_get_schema(ldb);
-        if (!schema) {
-                return WERR_DS_SCHEMA_NOT_LOADED;
-        }
-
-        status = dsdb_verify_oid_mappings_drsuapi(schema, mapping_ctr);
-        W_ERROR_NOT_OK_RETURN(status);
 
         out = talloc_zero(mem_ctx, struct dsdb_extended_replicated_objects);
@@ -237 +391 @@
         out->version            = DSDB_EXTENDED_REPLICATED_OBJECTS_VERSION;
 
-        out->partition_dn       = ldb_dn_new(out, ldb, partition_dn);
-        W_ERROR_HAVE_NO_MEMORY(out->partition_dn);
+        /*
+         * Ensure schema is kept valid for as long as 'out'
+         * which may contain pointers to it
+         */
+        schema = talloc_reference(out, schema);
+        W_ERROR_HAVE_NO_MEMORY(schema);
+
+        partition_dn = ldb_dn_new(out, ldb, partition_dn_str);
+        W_ERROR_HAVE_NO_MEMORY_AND_FREE(partition_dn, out);
+
+        status = dsdb_schema_pfm_from_drsuapi_pfm(mapping_ctr, true,
+                                                  out, &pfm_remote, NULL);
+        if (!W_ERROR_IS_OK(status)) {
+                DEBUG(0,(__location__ ": Failed to decode remote prefixMap: %s",
+                         win_errstr(status)));
+                talloc_free(out);
+                return status;
+        }
+
+        if (ldb_dn_compare(partition_dn, ldb_get_schema_basedn(ldb)) != 0) {
+                /*
+                 * check for schema changes in case
+                 * we are not replicating Schema NC
+                 */
+                status = dsdb_schema_info_cmp(schema, mapping_ctr);
+                if (!W_ERROR_IS_OK(status)) {
+                        DEBUG(1,("Remote schema has changed while replicating %s\n",
+                                 partition_dn_str));
+                        talloc_free(out);
+                        return status;
+                }
+        }
+
+        out->partition_dn       = partition_dn;
 
         out->source_dsa         = source_dsa;
@@ -247 +433 @@
                                                struct dsdb_extended_replicated_object,
                                                out->num_objects);
-        W_ERROR_HAVE_NO_MEMORY(out->objects);
+        W_ERROR_HAVE_NO_MEMORY_AND_FREE(out->objects, out);
 
         /* pass the linked attributes down to the repl_meta_data
@@ -256 +442 @@
         for (i=0, cur = first_object; cur; cur = cur->next_object, i++) {
                 if (i == out->num_objects) {
+                        talloc_free(out);
                         return WERR_FOOBAR;
                 }
 
-                status = dsdb_convert_object_ex(ldb, schema,
+                status = dsdb_convert_object_ex(ldb, schema, pfm_remote,
                                                 cur, gensec_skey,
+                                                NULL,
                                                 out->objects, &out->objects[i]);
                 if (!W_ERROR_IS_OK(status)) {
-                        DEBUG(0,("Failed to convert object %s\n", cur->object.identifier->dn));
+                        talloc_free(out);
+                        DEBUG(0,("Failed to convert object %s: %s\n",
+                                 cur->object.identifier->dn,
+                                 win_errstr(status)));
                         return status;
                 }
         }
         if (i != out->num_objects) {
-                return WERR_FOOBAR;
-        }
+                talloc_free(out);
+                return WERR_FOOBAR;
+        }
+
+        /* free pfm_remote, we won't need it anymore */
+        talloc_free(pfm_remote);
+
+        *objects = out;
+        return WERR_OK;
+}
+
+/**
+ * Commits a list of replicated objects.
+ *
+ * @param working_schema dsdb_schema to be used for resolving
+ *                       Classes/Attributes during Schema replication. If not NULL,
+ *                       it will be set on ldb and used while committing replicated objects
+ */
+WERROR dsdb_replicated_objects_commit(struct ldb_context *ldb,
+                                      struct dsdb_schema *working_schema,
+                                      struct dsdb_extended_replicated_objects *objects,
+                                      uint64_t *notify_uSN)
+{
+        WERROR werr;
+        struct ldb_result *ext_res;
+        struct dsdb_schema *cur_schema = NULL;
+        int ret;
+        uint64_t seq_num1, seq_num2;
 
         /* TODO: handle linked attributes */
@@ -279 +496 @@
         if (ret != LDB_SUCCESS) {
                 DEBUG(0,(__location__ " Failed to start transaction\n"));
-                talloc_free(out);
-                return WERR_FOOBAR;
-        }
-
-        ret = dsdb_load_partition_usn(ldb, out->partition_dn, &seq_num1);
+                return WERR_FOOBAR;
+        }
+
+        ret = dsdb_load_partition_usn(ldb, objects->partition_dn, &seq_num1, NULL);
         if (ret != LDB_SUCCESS) {
                 DEBUG(0,(__location__ " Failed to load partition uSN\n"));
-                talloc_free(out);
                 ldb_transaction_cancel(ldb);
                 return WERR_FOOBAR;             
         }
 
-        ret = ldb_extended(ldb, DSDB_EXTENDED_REPLICATED_OBJECTS_OID, out, &ext_res);
+        /*
+         * Set working_schema for ldb in case we are replicating from Schema NC.
+         * Schema won't be reloaded during Replicated Objects commit, as it is
+         * done in a transaction. So we need some way to search for newly
+         * added Classes and Attributes
+         */
+        if (working_schema) {
+                /* store current schema so we can fall back in case of failure */
+                cur_schema = dsdb_get_schema(ldb, working_schema);
+
+                ret = dsdb_reference_schema(ldb, working_schema, false);
+                if (ret != LDB_SUCCESS) {
+                        DEBUG(0,(__location__ "Failed to reference working schema - %s\n",
+                                 ldb_strerror(ret)));
+                        /* TODO: Map LDB Error to NTSTATUS? */
+                        ldb_transaction_cancel(ldb);
+                        return WERR_INTERNAL_ERROR;
+                }
+        }
+
+        ret = ldb_extended(ldb, DSDB_EXTENDED_REPLICATED_OBJECTS_OID, objects, &ext_res);
         if (ret != LDB_SUCCESS) {
+                /* restore previous schema */
+                if (cur_schema ) {
+                        dsdb_reference_schema(ldb, cur_schema, false);
+                        dsdb_make_schema_global(ldb, cur_schema);
+                }
+
                 DEBUG(0,("Failed to apply records: %s: %s\n",
                          ldb_errstring(ldb), ldb_strerror(ret)));
-                talloc_free(out);
                 ldb_transaction_cancel(ldb);
                 return WERR_FOOBAR;
         }
         talloc_free(ext_res);
+
+        /* Save our updated prefixMap */
+        if (working_schema) {
+                werr = dsdb_write_prefixes_from_schema_to_ldb(working_schema,
+                                                              ldb,
+                                                              working_schema);
+                if (!W_ERROR_IS_OK(werr)) {
+                        /* restore previous schema */
+                        if (cur_schema ) {
+                                dsdb_reference_schema(ldb, cur_schema, false);
+                                dsdb_make_schema_global(ldb, cur_schema);
+                        }
+                        DEBUG(0,("Failed to save updated prefixMap: %s\n",
+                                 win_errstr(werr)));
+                        return werr;
+                }
+        }
 
         ret = ldb_transaction_prepare_commit(ldb);
         if (ret != LDB_SUCCESS) {
-                DEBUG(0,(__location__ " Failed to prepare commit of transaction\n"));
-                talloc_free(out);
-                return WERR_FOOBAR;
-        }
-
-        ret = dsdb_load_partition_usn(ldb, out->partition_dn, &seq_num2);
+                /* restore previous schema */
+                if (cur_schema ) {
+                        dsdb_reference_schema(ldb, cur_schema, false);
+                        dsdb_make_schema_global(ldb, cur_schema);
+                }
+                DEBUG(0,(__location__ " Failed to prepare commit of transaction: %s\n",
+                         ldb_errstring(ldb)));
+                return WERR_FOOBAR;
+        }
+
+        ret = dsdb_load_partition_usn(ldb, objects->partition_dn, &seq_num2, NULL);
         if (ret != LDB_SUCCESS) {
+                /* restore previous schema */
+                if (cur_schema ) {
+                        dsdb_reference_schema(ldb, cur_schema, false);
+                        dsdb_make_schema_global(ldb, cur_schema);
+                }
                 DEBUG(0,(__location__ " Failed to load partition uSN\n"));
-                talloc_free(out);
                 ldb_transaction_cancel(ldb);
                 return WERR_FOOBAR;             
@@ -325 +591 @@
         ret = ldb_transaction_commit(ldb);
         if (ret != LDB_SUCCESS) {
+                /* restore previous schema */
+                if (cur_schema ) {
+                        dsdb_reference_schema(ldb, cur_schema, false);
+                        dsdb_make_schema_global(ldb, cur_schema);
+                }
                 DEBUG(0,(__location__ " Failed to commit transaction\n"));
-                talloc_free(out);
-                return WERR_FOOBAR;
-        }
-
+                return WERR_FOOBAR;
+        }
+
+        /*
+         * Reset the Schema used by ldb. This will lead to
+         * a schema cache being refreshed from database.
+         */
+        if (working_schema) {
+                cur_schema = dsdb_get_schema(ldb, NULL);
+                /* TODO: What we do in case dsdb_get_schema() fail?
+                 *       We can't fallback at this point anymore */
+                if (cur_schema) {
+                        dsdb_make_schema_global(ldb, cur_schema);
+                }
+        }
 
         DEBUG(2,("Replicated %u objects (%u linked attributes) for %s\n",
-                 out->num_objects, out->linked_attributes_count,
-                 ldb_dn_get_linearized(out->partition_dn)));
+                 objects->num_objects, objects->linked_attributes_count,
+                 ldb_dn_get_linearized(objects->partition_dn)));
                  
-
-        if (_out) {
-                *_out = out;
-        } else {
-                talloc_free(out);
-        }
-
         return WERR_OK;
 }
 
-static WERROR dsdb_convert_object(struct ldb_context *ldb,
-                                  const struct dsdb_schema *schema,
-                                  const struct drsuapi_DsReplicaObjectListItem *in,
-                                  TALLOC_CTX *mem_ctx,
-                                  struct ldb_message **_msg)
+static WERROR dsdb_origin_object_convert(struct ldb_context *ldb,
+                                         const struct dsdb_schema *schema,
+                                         const struct drsuapi_DsReplicaObjectListItem *in,
+                                         TALLOC_CTX *mem_ctx,
+                                         struct ldb_message **_msg)
 {
         WERROR status;
-        uint32_t i;
+        unsigned int i;
         struct ldb_message *msg;
 
@@ -381 +656 @@
                 e = &msg->elements[i];
 
-                status = dsdb_attribute_drsuapi_to_ldb(ldb, schema, a, msg->elements, e);
+                status = dsdb_attribute_drsuapi_to_ldb(ldb, schema, schema->prefixmap,
+                                                       a, msg->elements, e);
                 W_ERROR_NOT_OK_RETURN(status);
         }
@@ -412 +688 @@
         int ret;
 
-        schema = dsdb_get_schema(ldb);
+        for (cur = first_object; cur; cur = cur->next_object) {
+                num_objects++;
+        }
+
+        if (num_objects == 0) {
+                return WERR_OK;
+        }
+
+        ret = ldb_transaction_start(ldb);
+        if (ret != LDB_SUCCESS) {
+                return WERR_DS_INTERNAL_FAILURE;
+        }
+
+        objects = talloc_array(mem_ctx, struct ldb_message *,
+                               num_objects);
+        if (objects == NULL) {
+                status = WERR_NOMEM;
+                goto cancel;
+        }
+
+        schema = dsdb_get_schema(ldb, objects);
         if (!schema) {
                 return WERR_DS_SCHEMA_NOT_LOADED;
         }
 
-        for (cur = first_object; cur; cur = cur->next_object) {
-                num_objects++;
-        }
-
-        if (num_objects == 0) {
-                return WERR_OK;
-        }
-
-        objects = talloc_array(mem_ctx, struct ldb_message *,
-                               num_objects);
-        W_ERROR_HAVE_NO_MEMORY(objects);
-
         for (i=0, cur = first_object; cur; cur = cur->next_object, i++) {
-                status = dsdb_convert_object(ldb, schema,
-                                             cur, objects, &objects[i]);
-                W_ERROR_NOT_OK_RETURN(status);
+                status = dsdb_origin_object_convert(ldb, schema, cur,
+                                                    objects, &objects[i]);
+                if (!W_ERROR_IS_OK(status)) {
+                        goto cancel;
+                }
         }
 
@@ -438 +724 @@
                            struct drsuapi_DsReplicaObjectIdentifier2,
                            num_objects);
-        W_ERROR_HAVE_NO_MEMORY(objects);
+        if (ids == NULL) {
+                status = WERR_NOMEM;
+                goto cancel;
+        }
 
         for (i=0; i < num_objects; i++) {
                 struct dom_sid *sid = NULL;
+                struct ldb_request *add_req;
 
                 DEBUG(6,(__location__ ": adding %s\n", 
                          ldb_dn_get_linearized(objects[i]->dn)));
+
+                ret = ldb_build_add_req(&add_req,
+                                        ldb,
+                                        objects,
+                                        objects[i],
+                                        NULL,
+                                        NULL,
+                                        ldb_op_default_callback,
+                                        NULL);
+                if (ret != LDB_SUCCESS) {
+                        status = WERR_DS_INTERNAL_FAILURE;
+                        goto cancel;
+                }
+
+                ret = ldb_request_add_control(add_req, LDB_CONTROL_RELAX_OID, true, NULL);
+                if (ret != LDB_SUCCESS) {
+                        status = WERR_DS_INTERNAL_FAILURE;
+                        goto cancel;
+                }
                 
-                ret = ldb_add(ldb, objects[i]);
-                if (ret != 0) {
+                ret = ldb_request(ldb, add_req);
+                if (ret == LDB_SUCCESS) {
+                        ret = ldb_wait(add_req->handle, LDB_WAIT_ALL);
+                }
+                if (ret != LDB_SUCCESS) {
+                        DEBUG(0,(__location__ ": Failed add of %s - %s\n",
+                                 ldb_dn_get_linearized(objects[i]->dn), ldb_errstring(ldb)));
+                        status = WERR_DS_INTERNAL_FAILURE;
                         goto cancel;
                 }
+
+                talloc_free(add_req);
+
                 ret = ldb_search(ldb, objects, &res, objects[i]->dn,
                                  LDB_SCOPE_BASE, attrs,
                                  "(objectClass=*)");
-                if (ret != 0) {
+                if (ret != LDB_SUCCESS) {
+                        status = WERR_DS_INTERNAL_FAILURE;
                         goto cancel;
                 }
@@ -465 +784 @@
         }
 
+        ret = ldb_transaction_commit(ldb);
+        if (ret != LDB_SUCCESS) {
+                return WERR_DS_INTERNAL_FAILURE;
+        }
+
         talloc_free(objects);
 
@@ -470 +794 @@
         *_ids = ids;
         return WERR_OK;
+
 cancel:
         talloc_free(objects);
         ldb_transaction_cancel(ldb);
-        return WERR_FOOBAR;
+        return status;
 }