X-Git-Url: http://git.samba.org/?a=blobdiff_plain;f=source4%2Fdsdb%2Fsamdb%2Fldb_modules%2Fpartition.c;h=613cfa3a1112b266a53dcd3184e2f4b445bce3b8;hb=87f31510475c6debd56ff874130f4f5d48bef9a5;hp=4626eefa7c3a82033fe054e118fffa846765fd27;hpb=e62200e253a393ce0a427ff315267efbd2eaeeea;p=samba.git

diff --git a/source4/dsdb/samdb/ldb_modules/partition.c b/source4/dsdb/samdb/ldb_modules/partition.c
index 4626eefa7c3..613cfa3a111 100644
--- a/source4/dsdb/samdb/ldb_modules/partition.c
+++ b/source4/dsdb/samdb/ldb_modules/partition.c
@@ -39,11 +39,12 @@ struct part_request {
 struct partition_context {
 	struct ldb_module *module;
 	struct ldb_request *req;
-	bool got_success;
 
 	struct part_request *part_req;
-	int num_requests;
-	int finished_requests;
+	unsigned int num_requests;
+	unsigned int finished_requests;
+
+	const char **referrals;
 };
 
 static struct partition_context *partition_init_ctx(struct ldb_module *module, struct ldb_request *req)
@@ -63,60 +64,34 @@ static struct partition_context *partition_init_ctx(struct ldb_module *module, s
 }
 
 /*
- *    helper functions to call the next module in chain
- *    */
-
-static int partition_request(struct ldb_module *module, struct ldb_request *request)
+ * helper functions to call the next module in chain
+ */
+int partition_request(struct ldb_module *module, struct ldb_request *request)
 {
-	int ret;
-	switch (request->operation) {
-	case LDB_SEARCH:
-		PARTITION_FIND_OP(module, search);
-		ret = module->ops->search(module, request);
-		break;
-	case LDB_ADD:
-		PARTITION_FIND_OP(module, add);
-		ret = module->ops->add(module, request);
-		break;
-	case LDB_MODIFY:
-		PARTITION_FIND_OP(module, modify);
-		ret = module->ops->modify(module, request);
-		break;
-	case LDB_DELETE:
-		PARTITION_FIND_OP(module, del);
-		ret = module->ops->del(module, request);
-		break;
-	case LDB_RENAME:
-		PARTITION_FIND_OP(module, rename);
-		ret = module->ops->rename(module, request);
-		break;
-	case LDB_EXTENDED:
-		PARTITION_FIND_OP(module, extended);
-		ret = module->ops->extended(module, request);
-		break;
-	default:
-		PARTITION_FIND_OP(module, request);
-		ret = module->ops->request(module, request);
-		break;
-	}
-	if (ret == LDB_SUCCESS) {
-		return ret;
-	}
-	if (!ldb_errstring(ldb_module_get_ctx(module))) {
-		/* Set a default error string, to place the blame somewhere */
-		ldb_asprintf_errstring(ldb_module_get_ctx(module),
-					"error in module %s: %s (%d)",
-					module->ops->name,
-					ldb_strerror(ret), ret);
+	if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) { \
+		const struct dsdb_control_current_partition *partition = NULL;
+		struct ldb_control *partition_ctrl = ldb_request_get_control(request, DSDB_CONTROL_CURRENT_PARTITION_OID);
+		if (partition_ctrl) {
+			partition = talloc_get_type(partition_ctrl->data,
+						    struct dsdb_control_current_partition);
+		}
+
+		if (partition != NULL) {
+			ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_request() -> %s",
+				  ldb_dn_get_linearized(partition->dn));			
+		} else {
+			ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_request() -> (metadata partition)");
+		}
 	}
-	return ret;
+
+	return ldb_next_request(module, request);
 }
 
 static struct dsdb_partition *find_partition(struct partition_private_data *data,
 					     struct ldb_dn *dn,
 					     struct ldb_request *req)
 {
-	int i;
+	unsigned int i;
 	struct ldb_control *partition_ctrl;
 
 	/* see if the request has the partition DN specified in a
@@ -158,26 +133,41 @@ static int partition_req_callback(struct ldb_request *req,
 	struct partition_context *ac;
 	struct ldb_module *module;
 	struct ldb_request *nreq;
-	int ret, i;
+	int ret;
 	struct partition_private_data *data;
+	struct ldb_control *partition_ctrl;
 
 	ac = talloc_get_type(req->context, struct partition_context);
-	data = talloc_get_type(ac->module->private_data, struct partition_private_data);
+	data = talloc_get_type(ldb_module_get_private(ac->module), struct partition_private_data);
 
 	if (!ares) {
 		return ldb_module_done(ac->req, NULL, NULL,
 					LDB_ERR_OPERATIONS_ERROR);
 	}
 
-	if (ares->error != LDB_SUCCESS && !ac->got_success) {
+	partition_ctrl = ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID);
+	if (partition_ctrl && (ac->num_requests == 1 || ares->type == LDB_REPLY_ENTRY)) {
+		/* If we didn't fan this request out to mulitple partitions,
+		 * or this is an individual search result, we can
+		 * deterministily tell the caller what partition this was
+		 * written to (repl_meta_data likes to know) */
+		ret = ldb_reply_add_control(ares,
+					    DSDB_CONTROL_CURRENT_PARTITION_OID,
+					    false, partition_ctrl->data);
+		if (ret != LDB_SUCCESS) {
+			return ldb_module_done(ac->req, NULL, NULL,
+					       ret);
+		}
+	}
+
+	if (ares->error != LDB_SUCCESS) {
 		return ldb_module_done(ac->req, ares->controls,
 					ares->response, ares->error);
 	}
 
 	switch (ares->type) {
 	case LDB_REPLY_REFERRAL:
-		/* ignore referrals for now */
-		break;
+		return ldb_module_send_referral(ac->req, ares->referral);
 
 	case LDB_REPLY_ENTRY:
 		if (ac->req->operation != LDB_SEARCH) {
@@ -187,28 +177,10 @@ static int partition_req_callback(struct ldb_request *req,
 			return ldb_module_done(ac->req, NULL, NULL,
 						LDB_ERR_OPERATIONS_ERROR);
 		}
-		for (i=0; data && data->partitions && data->partitions[i]; i++) {
-			if (ldb_dn_compare(ares->message->dn, data->partitions[i]->ctrl->dn) == 0) {
-				struct ldb_control *part_control;
-				/* this is a partition root message - make
-				   sure it isn't one of our fake root
-				   entries from a parent partition */
-				part_control = ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID);
-				if (part_control && part_control->data != data->partitions[i]->ctrl) {
-					DEBUG(6,(__location__ ": Discarding partition mount object %s\n",
-						 ldb_dn_get_linearized(ares->message->dn)));
-					talloc_free(ares);
-					return LDB_SUCCESS;
-				}
-			}
-		}
 		
 		return ldb_module_send_entry(ac->req, ares->message, ares->controls);
 
 	case LDB_REPLY_DONE:
-		if (ares->error == LDB_SUCCESS) {
-			ac->got_success = true;
-		}
 		if (ac->req->operation == LDB_EXTENDED) {
 			/* FIXME: check for ares->response, replmd does not fill it ! */
 			if (ares->response) {
@@ -226,10 +198,23 @@ static int partition_req_callback(struct ldb_request *req,
 
 		ac->finished_requests++;
 		if (ac->finished_requests == ac->num_requests) {
+			/* Send back referrals if they do exist (search ops) */
+			if (ac->referrals != NULL) {
+				const char **ref;
+				for (ref = ac->referrals; *ref != NULL; ++ref) {
+					ret = ldb_module_send_referral(ac->req,
+								       talloc_strdup(ac->req, *ref));
+					if (ret != LDB_SUCCESS) {
+						return ldb_module_done(ac->req, NULL, NULL,
+								       ret);
+					}
+				}
+			}
+
 			/* this was the last one, call callback */
 			return ldb_module_done(ac->req, ares->controls,
 					       ares->response, 
-					       ac->got_success?LDB_SUCCESS:ares->error);
+					       ares->error);
 		}
 
 		/* not the last, now call the next one */
@@ -259,8 +244,7 @@ static int partition_prep_request(struct partition_context *ac,
 					struct part_request,
 					ac->num_requests + 1);
 	if (ac->part_req == NULL) {
-		ldb_oom(ldb_module_get_ctx(ac->module));
-		return LDB_ERR_OPERATIONS_ERROR;
+		return ldb_oom(ldb_module_get_ctx(ac->module));
 	}
 
 	switch (ac->req->operation) {
@@ -274,6 +258,7 @@ static int partition_prep_request(struct partition_context *ac,
 					ac->req->controls,
 					ac, partition_req_callback,
 					ac->req);
+		LDB_REQ_SET_LOCATION(req);
 		break;
 	case LDB_ADD:
 		ret = ldb_build_add_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
@@ -281,6 +266,7 @@ static int partition_prep_request(struct partition_context *ac,
 					ac->req->controls,
 					ac, partition_req_callback,
 					ac->req);
+		LDB_REQ_SET_LOCATION(req);
 		break;
 	case LDB_MODIFY:
 		ret = ldb_build_mod_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
@@ -288,6 +274,7 @@ static int partition_prep_request(struct partition_context *ac,
 					ac->req->controls,
 					ac, partition_req_callback,
 					ac->req);
+		LDB_REQ_SET_LOCATION(req);
 		break;
 	case LDB_DELETE:
 		ret = ldb_build_del_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
@@ -295,6 +282,7 @@ static int partition_prep_request(struct partition_context *ac,
 					ac->req->controls,
 					ac, partition_req_callback,
 					ac->req);
+		LDB_REQ_SET_LOCATION(req);
 		break;
 	case LDB_RENAME:
 		ret = ldb_build_rename_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
@@ -303,6 +291,7 @@ static int partition_prep_request(struct partition_context *ac,
 					ac->req->controls,
 					ac, partition_req_callback,
 					ac->req);
+		LDB_REQ_SET_LOCATION(req);
 		break;
 	case LDB_EXTENDED:
 		ret = ldb_build_extended_req(&req, ldb_module_get_ctx(ac->module),
@@ -312,6 +301,7 @@ static int partition_prep_request(struct partition_context *ac,
 					ac->req->controls,
 					ac, partition_req_callback,
 					ac->req);
+		LDB_REQ_SET_LOCATION(req);
 		break;
 	default:
 		ldb_set_errstring(ldb_module_get_ctx(ac->module),
@@ -329,8 +319,7 @@ static int partition_prep_request(struct partition_context *ac,
 		req->controls = talloc_memdup(req, ac->req->controls,
 					talloc_get_size(ac->req->controls));
 		if (req->controls == NULL) {
-			ldb_oom(ldb_module_get_ctx(ac->module));
-			return LDB_ERR_OPERATIONS_ERROR;
+			return ldb_oom(ldb_module_get_ctx(ac->module));
 		}
 	}
 
@@ -357,9 +346,9 @@ static int partition_prep_request(struct partition_context *ac,
 		}
 
 	} else {
-		/* make sure you put the NEXT module here, or
-		 * partition_request() will simply loop forever on itself */
-		ac->part_req[ac->num_requests].module = ac->module->next;
+		/* make sure you put the module here, or
+		 * or ldb_next_request() will skip a module */
+		ac->part_req[ac->num_requests].module = ac->module;
 	}
 
 	ac->num_requests++;
@@ -379,8 +368,8 @@ static int partition_send_all(struct ldb_module *module,
 			      struct partition_context *ac, 
 			      struct ldb_request *req) 
 {
-	int i;
-	struct partition_private_data *data = talloc_get_type(module->private_data, 
+	unsigned int i;
+	struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 							      struct partition_private_data);
 	int ret = partition_prep_request(ac, NULL);
 	if (ret != LDB_SUCCESS) {
@@ -404,21 +393,16 @@ static int partition_send_all(struct ldb_module *module,
 static int partition_replicate(struct ldb_module *module, struct ldb_request *req, struct ldb_dn *dn) 
 {
 	struct partition_context *ac;
-	unsigned i;
+	unsigned int i;
 	int ret;
 	struct dsdb_partition *partition;
-	struct partition_private_data *data = talloc_get_type(module->private_data, 
+	struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 							      struct partition_private_data);
 	if (!data || !data->partitions) {
 		return ldb_next_request(module, req);
 	}
 
-	ret = partition_reload_if_required(module, data);
-	if (ret != LDB_SUCCESS) {
-		return ret;
-	}
-	
-	if (req->operation != LDB_SEARCH) {
+	if (req->operation != LDB_SEARCH && ldb_dn_is_special(dn)) {
 		/* Is this a special DN, we need to replicate to every backend? */
 		for (i=0; data->replicate && data->replicate[i]; i++) {
 			if (ldb_dn_compare(data->replicate[i], 
@@ -426,7 +410,7 @@ static int partition_replicate(struct ldb_module *module, struct ldb_request *re
 				
 				ac = partition_init_ctx(module, req);
 				if (!ac) {
-					return LDB_ERR_OPERATIONS_ERROR;
+					return ldb_operr(ldb_module_get_ctx(module));
 				}
 				
 				return partition_send_all(module, ac, req);
@@ -452,7 +436,7 @@ static int partition_replicate(struct ldb_module *module, struct ldb_request *re
 
 	ac = partition_init_ctx(module, req);
 	if (!ac) {
-		return LDB_ERR_OPERATIONS_ERROR;
+		return ldb_operr(ldb_module_get_ctx(module));
 	}
 
 	/* we need to add a control but we never touch the original request */
@@ -468,24 +452,24 @@ static int partition_replicate(struct ldb_module *module, struct ldb_request *re
 /* search */
 static int partition_search(struct ldb_module *module, struct ldb_request *req)
 {
-	int ret;
 	struct ldb_control **saved_controls;
 	/* Find backend */
-	struct partition_private_data *data = talloc_get_type(module->private_data, 
+	struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 							      struct partition_private_data);
-
-	/* issue request */
-
-	/* (later) consider if we should be searching multiple
-	 * partitions (for 'invisible' partition behaviour */
+	struct partition_context *ac;
+	struct ldb_context *ldb;
+	struct loadparm_context *lp_ctx;
 
 	struct ldb_control *search_control = ldb_request_get_control(req, LDB_CONTROL_SEARCH_OPTIONS_OID);
 	struct ldb_control *domain_scope_control = ldb_request_get_control(req, LDB_CONTROL_DOMAIN_SCOPE_OID);
 	
 	struct ldb_search_options_control *search_options = NULL;
 	struct dsdb_partition *p;
+	unsigned int i, j;
+	int ret;
+	bool domain_scope = false, phantom_root = false;
 	
-	ret = partition_reload_if_required(module, data);
+	ret = partition_reload_if_required(module, data, req);
 	if (ret != LDB_SUCCESS) {
 		return ret;
 	}
@@ -495,114 +479,178 @@ static int partition_search(struct ldb_module *module, struct ldb_request *req)
 		/* the caller specified what partition they want the
 		 * search - just pass it on
 		 */
-		return ldb_next_request(p->module, req);		
+		return ldb_next_request(p->module, req);
 	}
 
-
+	/* Get back the search options from the search control, and mark it as
+	 * non-critical (to make backends and also dcpromo happy).
+	 */
 	if (search_control) {
 		search_options = talloc_get_type(search_control->data, struct ldb_search_options_control);
+		search_control->critical = 0;
+
 	}
 
-	/* Remove the domain_scope control, so we don't confuse a backend server */
-	if (domain_scope_control && !save_controls(domain_scope_control, req, &saved_controls)) {
-		ldb_oom(ldb_module_get_ctx(module));
-		return LDB_ERR_OPERATIONS_ERROR;
+	/* Remove the "domain_scope" control, so we don't confuse a backend
+	 * server */
+	if (domain_scope_control && !ldb_save_controls(domain_scope_control, req, &saved_controls)) {
+		return ldb_oom(ldb_module_get_ctx(module));
 	}
 
-	/*
-	 * for now pass down the LDB_CONTROL_SEARCH_OPTIONS_OID control
-	 * down as uncritical to make windows 2008 dcpromo happy.
-	 */
-	if (search_control) {
-		search_control->critical = 0;
+	/* Locate the options */
+	domain_scope = (search_options
+		&& (search_options->search_options & LDB_SEARCH_OPTION_DOMAIN_SCOPE))
+		|| domain_scope_control;
+	phantom_root = search_options
+		&& (search_options->search_options & LDB_SEARCH_OPTION_PHANTOM_ROOT);
+
+	/* Remove handled options from the search control flag */
+	if (search_options) {
+		search_options->search_options = search_options->search_options
+			& ~LDB_SEARCH_OPTION_DOMAIN_SCOPE
+			& ~LDB_SEARCH_OPTION_PHANTOM_ROOT;
 	}
 
-	/* TODO:
-	   Generate referrals (look for a partition under this DN) if we don't have the above control specified
-	*/
-	
-	if (search_options && (search_options->search_options & LDB_SEARCH_OPTION_PHANTOM_ROOT)) {
-		int i;
-		struct partition_context *ac;
-		if ((search_options->search_options & ~LDB_SEARCH_OPTION_PHANTOM_ROOT) == 0) {
-			/* We have processed this flag, so we are done with this control now */
-
-			/* Remove search control, so we don't confuse a backend server */
-			if (search_control && !save_controls(search_control, req, &saved_controls)) {
-				ldb_oom(ldb_module_get_ctx(module));
-				return LDB_ERR_OPERATIONS_ERROR;
-			}
-		}
-		ac = partition_init_ctx(module, req);
-		if (!ac) {
-			return LDB_ERR_OPERATIONS_ERROR;
-		}
+	if (!data || !data->partitions) {
+		return ldb_next_request(module, req);
+	}
 
-		/* Search from the base DN */
-		if (!req->op.search.base || ldb_dn_is_null(req->op.search.base)) {
-			return partition_send_all(module, ac, req);
-		}
-		for (i=0; data && data->partitions && data->partitions[i]; i++) {
-			bool match = false, stop = false;
-			/* Find all partitions under the search base 
-			   
-			   we match if:
-
-			      1) the DN we are looking for exactly matches the partition
-		             or
-			      2) the DN we are looking for is a parent of the partition and it isn't
-                                 a scope base search
-                             or
-			      3) the DN we are looking for is a child of the partition
+	ac = partition_init_ctx(module, req);
+	if (!ac) {
+		return ldb_operr(ldb_module_get_ctx(module));
+	}
+
+	ldb = ldb_module_get_ctx(ac->module);
+	lp_ctx = talloc_get_type(ldb_get_opaque(ldb, "loadparm"),
+						struct loadparm_context);
+
+	/* Search from the base DN */
+	if (ldb_dn_is_null(req->op.search.base)) {
+		return partition_send_all(module, ac, req);
+	}
+
+	for (i=0; data->partitions[i]; i++) {
+		bool match = false, stop = false;
+
+		if (phantom_root) {
+			/* Phantom root: Find all partitions under the
+			 * search base. We match if:
+			 *
+			 * 1) the DN we are looking for exactly matches a
+			 *    certain partition and always stop
+			 * 2) the DN we are looking for is a parent of certain
+			 *    partitions and it isn't a scope base search
+			 * 3) the DN we are looking for is a child of a certain
+			 *    partition and always stop
+			 *    - we don't need to go any further up in the
+			 *    hierarchy!
 			 */
-			if (ldb_dn_compare(data->partitions[i]->ctrl->dn, req->op.search.base) == 0) {
+			if (ldb_dn_compare(data->partitions[i]->ctrl->dn,
+					   req->op.search.base) == 0) {
 				match = true;
-				if (req->op.search.scope == LDB_SCOPE_BASE) {
-					stop = true;
-				}
+				stop = true;
 			}
-			if (!match && 
-			    (ldb_dn_compare_base(req->op.search.base, data->partitions[i]->ctrl->dn) == 0 &&
+			if (!match &&
+			    (ldb_dn_compare_base(req->op.search.base,
+						 data->partitions[i]->ctrl->dn) == 0 &&
 			     req->op.search.scope != LDB_SCOPE_BASE)) {
 				match = true;
 			}
 			if (!match &&
-			    ldb_dn_compare_base(data->partitions[i]->ctrl->dn, req->op.search.base) == 0) {
+			    ldb_dn_compare_base(data->partitions[i]->ctrl->dn,
+						req->op.search.base) == 0) {
 				match = true;
 				stop = true; /* note that this relies on partition ordering */
 			}
-			if (match) {
-				ret = partition_prep_request(ac, data->partitions[i]);
-				if (ret != LDB_SUCCESS) {
-					return ret;
+		} else {
+			/* Domain scope: Find all partitions under the search
+			 * base.
+			 *
+			 * We generate referral candidates if we haven't
+			 * specified the domain scope control, haven't a base
+			 * search* scope and the DN we are looking for is a real
+			 * predecessor of certain partitions. When a new
+			 * referral candidate is nearer to the DN than an
+			 * existing one delete the latter (we want to have only
+			 * the closest ones). When we checked this for all
+			 * candidates we have the final referrals.
+			 *
+			 * We match if the DN we are looking for is a child of
+			 * a certain partition or the partition
+			 * DN itself - we don't need to go any further
+			 * up in the hierarchy!
+			 */
+			if ((!domain_scope) &&
+			    (req->op.search.scope != LDB_SCOPE_BASE) &&
+			    (ldb_dn_compare_base(req->op.search.base,
+						 data->partitions[i]->ctrl->dn) == 0) &&
+			    (ldb_dn_compare(req->op.search.base,
+					    data->partitions[i]->ctrl->dn) != 0)) {
+				char *ref = talloc_asprintf(ac,
+							    "ldap://%s/%s%s",
+							    lpcfg_dnsdomain(lp_ctx),
+							    ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
+							    req->op.search.scope == LDB_SCOPE_ONELEVEL ? "??base" : "");
+
+				if (ref == NULL) {
+					return ldb_oom(ldb);
 				}
-			}
-			if (stop) break;
-		}
 
-		/* Perhaps we didn't match any partitions.  Try the main partition, only */
-		if (ac->num_requests == 0) {
-			talloc_free(ac);
-			return ldb_next_request(module, req);
-		}
+				/* Initialise the referrals list */
+				if (ac->referrals == NULL) {
+					ac->referrals = (const char **) str_list_make_empty(ac);
+					if (ac->referrals == NULL) {
+						return ldb_oom(ldb);
+					}
+				}
 
-		/* fire the first one */
-		return partition_call_first(ac);
+				/* Check if the new referral candidate is
+				 * closer to the base DN than already
+				 * saved ones and delete the latters */
+				j = 0;
+				while (ac->referrals[j] != NULL) {
+					if (strstr(ac->referrals[j],
+						   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn)) != NULL) {
+						str_list_remove(ac->referrals,
+								ac->referrals[j]);
+					} else {
+						++j;
+					}
+				}
 
-	} else {
-		/* Handle this like all other requests */
-		if (search_control && (search_options->search_options & ~LDB_SEARCH_OPTION_PHANTOM_ROOT) == 0) {
-			/* We have processed this flag, so we are done with this control now */
-
-			/* Remove search control, so we don't confuse a backend server */
-			if (search_control && !save_controls(search_control, req, &saved_controls)) {
-				ldb_oom(ldb_module_get_ctx(module));
-				return LDB_ERR_OPERATIONS_ERROR;
+				/* Add our new candidate */
+				ac->referrals = str_list_add(ac->referrals, ref);
+
+				talloc_free(ref);
+
+				if (ac->referrals == NULL) {
+					return ldb_oom(ldb);
+				}
+			}
+			if (ldb_dn_compare_base(data->partitions[i]->ctrl->dn, req->op.search.base) == 0) {
+				match = true;
+				stop = true; /* note that this relies on partition ordering */
+			}
+		}
+
+		if (match) {
+			ret = partition_prep_request(ac, data->partitions[i]);
+			if (ret != LDB_SUCCESS) {
+				return ret;
 			}
 		}
 
-		return partition_replicate(module, req, req->op.search.base);
+		if (stop) break;
 	}
+
+	/* Perhaps we didn't match any partitions. Try the main partition */
+	if (ac->num_requests == 0) {
+		talloc_free(ac);
+		return ldb_next_request(module, req);
+	}
+
+	/* fire the first one */
+	return partition_call_first(ac);
 }
 
 /* add */
@@ -626,21 +674,15 @@ static int partition_delete(struct ldb_module *module, struct ldb_request *req)
 /* rename */
 static int partition_rename(struct ldb_module *module, struct ldb_request *req)
 {
-	int ret;
 	/* Find backend */
 	struct dsdb_partition *backend, *backend2;
 	
-	struct partition_private_data *data = talloc_get_type(module->private_data, 
+	struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 							      struct partition_private_data);
 
 	/* Skip the lot if 'data' isn't here yet (initialisation) */
 	if (!data) {
-		return LDB_ERR_OPERATIONS_ERROR;
-	}
-
-	ret = partition_reload_if_required(module, data);
-	if (ret != LDB_SUCCESS) {
-		return ret;
+		return ldb_operr(ldb_module_get_ctx(module));
 	}
 
 	backend = find_partition(data, req->op.rename.olddn, req);
@@ -667,59 +709,73 @@ static int partition_rename(struct ldb_module *module, struct ldb_request *req)
 /* start a transaction */
 static int partition_start_trans(struct ldb_module *module)
 {
-	int i, ret;
-	struct partition_private_data *data = talloc_get_type(module->private_data, 
+	unsigned int i;
+	int ret;
+	struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 							      struct partition_private_data);
 	/* Look at base DN */
 	/* Figure out which partition it is under */
 	/* Skip the lot if 'data' isn't here yet (initialization) */
+	if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
+		ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_start_trans() -> (metadata partition)");
+	}
 	ret = ldb_next_start_trans(module);
 	if (ret != LDB_SUCCESS) {
 		return ret;
 	}
 
-	for (i=0; data && data->partitions && data->partitions[i]; i++) {
-		struct ldb_module *next = data->partitions[i]->module;
-		PARTITION_FIND_OP(next, start_transaction);
+	ret = partition_reload_if_required(module, data, NULL);
+	if (ret != LDB_SUCCESS) {
+		return ret;
+	}
 
-		ret = next->ops->start_transaction(next);
+	for (i=0; data && data->partitions && data->partitions[i]; i++) {
+		if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
+			ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_start_trans() -> %s",
+				  ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
+		}
+		ret = ldb_next_start_trans(data->partitions[i]->module);
 		if (ret != LDB_SUCCESS) {
 			/* Back it out, if it fails on one */
 			for (i--; i >= 0; i--) {
-				next = data->partitions[i]->module;
-				PARTITION_FIND_OP(next, del_transaction);
-
-				next->ops->del_transaction(next);
+				ldb_next_del_trans(data->partitions[i]->module);
 			}
 			ldb_next_del_trans(module);
 			return ret;
 		}
 	}
+
+	data->in_transaction++;
+
 	return LDB_SUCCESS;
 }
 
 /* prepare for a commit */
 static int partition_prepare_commit(struct ldb_module *module)
 {
-	int i;
-	struct partition_private_data *data = talloc_get_type(module->private_data, 
+	unsigned int i;
+	struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 							      struct partition_private_data);
 
 	for (i=0; data && data->partitions && data->partitions[i]; i++) {
-		struct ldb_module *next_prepare = data->partitions[i]->module;
 		int ret;
 
-		PARTITION_FIND_OP_NOERROR(next_prepare, prepare_commit);
-		if (next_prepare == NULL) {
-			continue;
+		if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
+			ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_prepare_commit() -> %s",
+				  ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
 		}
-
-		ret = next_prepare->ops->prepare_commit(next_prepare);
+		ret = ldb_next_prepare_commit(data->partitions[i]->module);
 		if (ret != LDB_SUCCESS) {
+			ldb_asprintf_errstring(ldb_module_get_ctx(module), "prepare_commit error on %s: %s",
+					       ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
+					       ldb_errstring(ldb_module_get_ctx(module)));
 			return ret;
 		}
 	}
 
+	if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
+		ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_prepare_commit() -> (metadata partition)");
+	}
 	return ldb_next_prepare_commit(module);
 }
 
@@ -727,40 +783,74 @@ static int partition_prepare_commit(struct ldb_module *module)
 /* end a transaction */
 static int partition_end_trans(struct ldb_module *module)
 {
-	int i;
-	struct partition_private_data *data = talloc_get_type(module->private_data, 
+	int ret, ret2;
+	unsigned int i;
+	struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 							      struct partition_private_data);
-	for (i=0; data && data->partitions && data->partitions[i]; i++) {
-		struct ldb_module *next_end = data->partitions[i]->module;
-		int ret;
 
-		PARTITION_FIND_OP(next_end, end_transaction);
+	ret = LDB_SUCCESS;
 
-		ret = next_end->ops->end_transaction(next_end);
-		if (ret != LDB_SUCCESS) {
-			return ret;
+	if (data->in_transaction == 0) {
+		DEBUG(0,("partition end transaction mismatch\n"));
+		ret = LDB_ERR_OPERATIONS_ERROR;
+	} else {
+		data->in_transaction--;
+	}
+
+	for (i=0; data && data->partitions && data->partitions[i]; i++) {
+		if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
+			ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_end_trans() -> %s",
+				  ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
+		}
+		ret2 = ldb_next_end_trans(data->partitions[i]->module);
+		if (ret2 != LDB_SUCCESS) {
+			ldb_asprintf_errstring(ldb_module_get_ctx(module), "end_trans error on %s: %s",
+					       ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
+					       ldb_errstring(ldb_module_get_ctx(module)));
+			ret = ret2;
 		}
 	}
 
-	return ldb_next_end_trans(module);
+	if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
+		ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_end_trans() -> (metadata partition)");
+	}
+	ret2 = ldb_next_end_trans(module);
+	if (ret2 != LDB_SUCCESS) {
+		ret = ret2;
+	}
+	return ret;
 }
 
 /* delete a transaction */
 static int partition_del_trans(struct ldb_module *module)
 {
-	int i, ret, final_ret = LDB_SUCCESS;
-	struct partition_private_data *data = talloc_get_type(module->private_data, 
+	int ret, final_ret = LDB_SUCCESS;
+	unsigned int i;
+	struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 							      struct partition_private_data);
 	for (i=0; data && data->partitions && data->partitions[i]; i++) {
-		struct ldb_module *next = data->partitions[i]->module;
-		PARTITION_FIND_OP(next, del_transaction);
-
-		ret = next->ops->del_transaction(next);
+		if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
+			ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_del_trans() -> %s",
+				  ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
+		}
+		ret = ldb_next_del_trans(data->partitions[i]->module);
 		if (ret != LDB_SUCCESS) {
+			ldb_asprintf_errstring(ldb_module_get_ctx(module), "del_trans error on %s: %s",
+					       ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
+					       ldb_errstring(ldb_module_get_ctx(module)));
 			final_ret = ret;
 		}
 	}	
 
+	if (data->in_transaction == 0) {
+		DEBUG(0,("partition del transaction mismatch\n"));
+		return ldb_operr(ldb_module_get_ctx(module));
+	}
+	data->in_transaction--;
+
+	if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
+		ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_del_trans() -> (metadata partition)");
+	}
 	ret = ldb_next_del_trans(module);
 	if (ret != LDB_SUCCESS) {
 		final_ret = ret;
@@ -778,12 +868,12 @@ int partition_primary_sequence_number(struct ldb_module *module, TALLOC_CTX *mem
 	struct ldb_seqnum_result *seqr;
 	res = talloc_zero(mem_ctx, struct ldb_result);
 	if (res == NULL) {
-		return LDB_ERR_OPERATIONS_ERROR;
+		return ldb_oom(ldb_module_get_ctx(module));
 	}
 	tseq = talloc_zero(res, struct ldb_seqnum_request);
 	if (tseq == NULL) {
 		talloc_free(res);
-		return LDB_ERR_OPERATIONS_ERROR;
+		return ldb_oom(ldb_module_get_ctx(module));
 	}
 	tseq->type = type;
 	
@@ -794,6 +884,7 @@ int partition_primary_sequence_number(struct ldb_module *module, TALLOC_CTX *mem
 				     res,
 				     ldb_extended_default_callback,
 				     NULL);
+	LDB_REQ_SET_LOCATION(treq);
 	if (ret != LDB_SUCCESS) {
 		talloc_free(res);
 		return ret;
@@ -827,11 +918,12 @@ int partition_primary_sequence_number(struct ldb_module *module, TALLOC_CTX *mem
 /* FIXME: This function is still semi-async */
 static int partition_sequence_number(struct ldb_module *module, struct ldb_request *req)
 {
-	int i, ret;
+	int ret;
+	unsigned int i;
 	uint64_t seq_number = 0;
 	uint64_t timestamp_sequence = 0;
 	uint64_t timestamp = 0;
-	struct partition_private_data *data = talloc_get_type(module->private_data, 
+	struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 							      struct partition_private_data);
 	struct ldb_seqnum_request *seq;
 	struct ldb_seqnum_result *seqr;
@@ -866,12 +958,12 @@ static int partition_sequence_number(struct ldb_module *module, struct ldb_reque
 
 			res = talloc_zero(req, struct ldb_result);
 			if (res == NULL) {
-				return LDB_ERR_OPERATIONS_ERROR;
+				return ldb_oom(ldb_module_get_ctx(module));
 			}
 			tseq = talloc_zero(res, struct ldb_seqnum_request);
 			if (tseq == NULL) {
 				talloc_free(res);
-				return LDB_ERR_OPERATIONS_ERROR;
+				return ldb_oom(ldb_module_get_ctx(module));
 			}
 			tseq->type = seq->type;
 
@@ -882,6 +974,7 @@ static int partition_sequence_number(struct ldb_module *module, struct ldb_reque
 						     res,
 						     ldb_extended_default_callback,
 						     NULL);
+			LDB_REQ_SET_LOCATION(treq);
 			if (ret != LDB_SUCCESS) {
 				talloc_free(res);
 				return ret;
@@ -922,13 +1015,13 @@ static int partition_sequence_number(struct ldb_module *module, struct ldb_reque
 
 		res = talloc_zero(req, struct ldb_result);
 		if (res == NULL) {
-			return LDB_ERR_OPERATIONS_ERROR;
+			return ldb_oom(ldb_module_get_ctx(module));
 		}
 
 		tseq = talloc_zero(res, struct ldb_seqnum_request);
 		if (tseq == NULL) {
 			talloc_free(res);
-			return LDB_ERR_OPERATIONS_ERROR;
+			return ldb_oom(ldb_module_get_ctx(module));
 		}
 		tseq->type = LDB_SEQ_HIGHEST_TIMESTAMP;
 
@@ -939,6 +1032,7 @@ static int partition_sequence_number(struct ldb_module *module, struct ldb_reque
 					     res,
 					     ldb_extended_default_callback,
 					     NULL);
+		LDB_REQ_SET_LOCATION(treq);
 		if (ret != LDB_SUCCESS) {
 			talloc_free(res);
 			return ret;
@@ -966,13 +1060,13 @@ static int partition_sequence_number(struct ldb_module *module, struct ldb_reque
 
 			res = talloc_zero(req, struct ldb_result);
 			if (res == NULL) {
-				return LDB_ERR_OPERATIONS_ERROR;
+				return ldb_oom(ldb_module_get_ctx(module));
 			}
 
 			tseq = talloc_zero(res, struct ldb_seqnum_request);
 			if (tseq == NULL) {
 				talloc_free(res);
-				return LDB_ERR_OPERATIONS_ERROR;
+				return ldb_oom(ldb_module_get_ctx(module));
 			}
 			tseq->type = LDB_SEQ_HIGHEST_TIMESTAMP;
 
@@ -983,6 +1077,7 @@ static int partition_sequence_number(struct ldb_module *module, struct ldb_reque
 						     res,
 						     ldb_extended_default_callback,
 						     NULL);
+			LDB_REQ_SET_LOCATION(treq);
 			if (ret != LDB_SUCCESS) {
 				talloc_free(res);
 				return ret;
@@ -1021,12 +1116,12 @@ static int partition_sequence_number(struct ldb_module *module, struct ldb_reque
 
 	ext = talloc_zero(req, struct ldb_extended);
 	if (!ext) {
-		return LDB_ERR_OPERATIONS_ERROR;
+		return ldb_oom(ldb_module_get_ctx(module));
 	}
 	seqr = talloc_zero(ext, struct ldb_seqnum_result);
 	if (seqr == NULL) {
 		talloc_free(ext);
-		return LDB_ERR_OPERATIONS_ERROR;
+		return ldb_oom(ldb_module_get_ctx(module));
 	}
 	ext->oid = LDB_EXTENDED_SEQUENCE_NUMBER;
 	ext->data = seqr;
@@ -1062,52 +1157,6 @@ static int partition_sequence_number(struct ldb_module *module, struct ldb_reque
 	return ldb_module_done(req, NULL, ext, LDB_SUCCESS);
 }
 
-static int partition_extended_schema_update_now(struct ldb_module *module, struct ldb_request *req)
-{
-	struct dsdb_partition *partition;
-	struct partition_private_data *data;
-	struct ldb_dn *schema_dn;
-	struct partition_context *ac;
-	int ret;
-
-	schema_dn = talloc_get_type(req->op.extended.data, struct ldb_dn);
-	if (!schema_dn) {
-		ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_FATAL, "partition_extended: invalid extended data\n");
-		return LDB_ERR_PROTOCOL_ERROR;
-	}
-
-	data = talloc_get_type(module->private_data, struct partition_private_data);
-	if (!data) {
-		return LDB_ERR_OPERATIONS_ERROR;
-	}
-	
-	partition = find_partition( data, schema_dn, req);
-	if (!partition) {
-		return ldb_next_request(module, req);
-	}
-
-	ac = partition_init_ctx(module, req);
-	if (!ac) {
-		return LDB_ERR_OPERATIONS_ERROR;
-	}
-
-	/* we need to add a control but we never touch the original request */
-	ret = partition_prep_request(ac, partition);
-	if (ret != LDB_SUCCESS) {
-		return ret;
-	}
-
-	/* fire the first one */
-	ret = partition_call_first(ac);
-
-	if (ret != LDB_SUCCESS){
-		return ret;
-	}
-
-	return ldb_request_done(req, ret);
-}
-
-
 /* extended */
 static int partition_extended(struct ldb_module *module, struct ldb_request *req)
 {
@@ -1115,12 +1164,12 @@ static int partition_extended(struct ldb_module *module, struct ldb_request *req
 	struct partition_context *ac;
 	int ret;
 
-	data = talloc_get_type(module->private_data, struct partition_private_data);
+	data = talloc_get_type(ldb_module_get_private(module), struct partition_private_data);
 	if (!data) {
 		return ldb_next_request(module, req);
 	}
 
-	ret = partition_reload_if_required(module, data);
+	ret = partition_reload_if_required(module, data, req);
 	if (ret != LDB_SUCCESS) {
 		return ret;
 	}
@@ -1133,11 +1182,6 @@ static int partition_extended(struct ldb_module *module, struct ldb_request *req
 		return partition_create(module, req);
 	}
 
-	/* forward schemaUpdateNow operation to schema_fsmo module*/
-	if (strcmp(req->op.extended.oid, DSDB_EXTENDED_SCHEMA_UPDATE_NOW_OID) == 0) {
-		return partition_extended_schema_update_now( module, req );
-	}	
-
 	/* 
 	 * as the extended operation has no dn
 	 * we need to send it to all partitions
@@ -1145,13 +1189,13 @@ static int partition_extended(struct ldb_module *module, struct ldb_request *req
 
 	ac = partition_init_ctx(module, req);
 	if (!ac) {
-		return LDB_ERR_OPERATIONS_ERROR;
+		return ldb_operr(ldb_module_get_ctx(module));
 	}
 
 	return partition_send_all(module, ac, req);
 }
 
-_PUBLIC_ const struct ldb_module_ops ldb_partition_module_ops = {
+static const struct ldb_module_ops ldb_partition_module_ops = {
 	.name		   = "partition",
 	.init_context	   = partition_init,
 	.search            = partition_search,
@@ -1165,3 +1209,9 @@ _PUBLIC_ const struct ldb_module_ops ldb_partition_module_ops = {
 	.end_transaction   = partition_end_trans,
 	.del_transaction   = partition_del_trans,
 };
+
+int ldb_partition_module_init(const char *version)
+{
+	LDB_MODULE_CHECK_VERSION(version);
+	return ldb_register_module(&ldb_partition_module_ops);
+}