repl_meta_data.c

samba最新软件

				continue;
			}

			found = true;

			/*
			 * we update only the highest_usn and not the latest_sync_success time,
			 * because the last success stands for direct replication
			 */
			if (ruv->cursors[i].highest_usn > nuv.ctr.ctr2.cursors[j].highest_usn) {
				nuv.ctr.ctr2.cursors[j].highest_usn = ruv->cursors[i].highest_usn;
			}
			break;			
		}

		if (found) continue;

		/* if it's not there yet, add it */
		nuv.ctr.ctr2.cursors[ni] = ruv->cursors[i];
		ni++;
	}

	/*
	 * merge in the current highwatermark for the source_dsa
	 */
	found = false;
	for (j=0; j < ni; j++) {
		if (!GUID_equal(&ar->objs->source_dsa->source_dsa_invocation_id,
				&nuv.ctr.ctr2.cursors[j].source_dsa_invocation_id)) {
			continue;
		}

		found = true;

		/*
		 * here we update the highest_usn and last_sync_success time
		 * because we're directly replicating from the source_dsa
		 *
		 * and use the tmp_highest_usn because this is what we have just applied
		 * to our ldb
		 */
		nuv.ctr.ctr2.cursors[j].highest_usn		= ar->objs->source_dsa->highwatermark.tmp_highest_usn;
		nuv.ctr.ctr2.cursors[j].last_sync_success	= now;
		break;
	}
	if (!found) {
		/*
		 * here we update the highest_usn and last_sync_success time
		 * because we're directly replicating from the source_dsa
		 *
		 * and use the tmp_highest_usn because this is what we have just applied
		 * to our ldb
		 */
		nuv.ctr.ctr2.cursors[ni].source_dsa_invocation_id= ar->objs->source_dsa->source_dsa_invocation_id;
		nuv.ctr.ctr2.cursors[ni].highest_usn		= ar->objs->source_dsa->highwatermark.tmp_highest_usn;
		nuv.ctr.ctr2.cursors[ni].last_sync_success	= now;
		ni++;
	}

	/*
	 * finally correct the size of the cursors array
	 */
	nuv.ctr.ctr2.count = ni;

	/*
	 * sort the cursors
	 */
	qsort(nuv.ctr.ctr2.cursors, nuv.ctr.ctr2.count,
	      sizeof(struct drsuapi_DsReplicaCursor2),
	      (comparison_fn_t)replmd_drsuapi_DsReplicaCursor2_compare);

	/*
	 * create the change ldb_message
	 */
	msg = ldb_msg_new(ar->sub.mem_ctx);
	if (!msg) return replmd_replicated_request_werror(ar, WERR_NOMEM);
	msg->dn = ar->sub.search_msg->dn;

	ndr_err = ndr_push_struct_blob(&nuv_value, msg, 
				       lp_iconv_convenience(ldb_get_opaque(ar->module->ldb, "loadparm")), 
				       &nuv,
				       (ndr_push_flags_fn_t)ndr_push_replUpToDateVectorBlob);
	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
		NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
		return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
	}
	ret = ldb_msg_add_value(msg, "replUpToDateVector", &nuv_value, &nuv_el);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}
	nuv_el->flags = LDB_FLAG_MOD_REPLACE;

	/*
	 * now create the new repsFrom value from the given repsFromTo1 structure
	 */
	ZERO_STRUCT(nrf);
	nrf.version					= 1;
	nrf.ctr.ctr1					= *ar->objs->source_dsa;
	/* and fix some values... */
	nrf.ctr.ctr1.consecutive_sync_failures		= 0;
	nrf.ctr.ctr1.last_success			= now;
	nrf.ctr.ctr1.last_attempt			= now;
	nrf.ctr.ctr1.result_last_attempt		= WERR_OK;
	nrf.ctr.ctr1.highwatermark.highest_usn		= nrf.ctr.ctr1.highwatermark.tmp_highest_usn;

	/*
	 * first see if we already have a repsFrom value for the current source dsa
	 * if so we'll later replace this value
	 */
	orf_el = ldb_msg_find_element(ar->sub.search_msg, "repsFrom");
	if (orf_el) {
		for (i=0; i < orf_el->num_values; i++) {
			struct repsFromToBlob *trf;

			trf = talloc(ar->sub.mem_ctx, struct repsFromToBlob);
			if (!trf) return replmd_replicated_request_werror(ar, WERR_NOMEM);

			ndr_err = ndr_pull_struct_blob(&orf_el->values[i], trf, lp_iconv_convenience(ldb_get_opaque(ar->module->ldb, "loadparm")), trf,
						       (ndr_pull_flags_fn_t)ndr_pull_repsFromToBlob);
			if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
				NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
				return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
			}

			if (trf->version != 1) {
				return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
			}

			/*
			 * we compare the source dsa objectGUID not the invocation_id
			 * because we want only one repsFrom value per source dsa
			 * and when the invocation_id of the source dsa has changed we don't need 
			 * the old repsFrom with the old invocation_id
			 */
			if (!GUID_equal(&trf->ctr.ctr1.source_dsa_obj_guid,
					&ar->objs->source_dsa->source_dsa_obj_guid)) {
				talloc_free(trf);
				continue;
			}

			talloc_free(trf);
			nrf_value = &orf_el->values[i];
			break;
		}

		/*
		 * copy over all old values to the new ldb_message
		 */
		ret = ldb_msg_add_empty(msg, "repsFrom", 0, &nrf_el);
		if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);
		*nrf_el = *orf_el;
	}

	/*
	 * if we haven't found an old repsFrom value for the current source dsa
	 * we'll add a new value
	 */
	if (!nrf_value) {
		struct ldb_val zero_value;
		ZERO_STRUCT(zero_value);
		ret = ldb_msg_add_value(msg, "repsFrom", &zero_value, &nrf_el);
		if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);

		nrf_value = &nrf_el->values[nrf_el->num_values - 1];
	}

	/* we now fill the value which is already attached to ldb_message */
	ndr_err = ndr_push_struct_blob(nrf_value, msg, 
				       lp_iconv_convenience(ldb_get_opaque(ar->module->ldb, "loadparm")),
				       &nrf,
				       (ndr_push_flags_fn_t)ndr_push_repsFromToBlob);
	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
		NTSTATUS nt_status = ndr_map_error2ntstatus(ndr_err);
		return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
	}

	/* 
	 * the ldb_message_element for the attribute, has all the old values and the new one
	 * so we'll replace the whole attribute with all values
	 */
	nrf_el->flags = LDB_FLAG_MOD_REPLACE;

	/* prepare the ldb_modify() request */
	ret = ldb_build_mod_req(&ar->sub.change_req,
				ar->module->ldb,
				ar->sub.mem_ctx,
				msg,
				NULL,
				ar,
				replmd_replicated_uptodate_modify_callback);
	if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);

#ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */ 
	return ldb_next_request(ar->module, ar->sub.change_req);
#else
	ret = ldb_next_request(ar->module, ar->sub.change_req);
	if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);

	ar->sub.change_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
	if (ar->sub.change_ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ar->sub.change_ret);
	}

	talloc_free(ar->sub.mem_ctx);
	ZERO_STRUCT(ar->sub);

	return replmd_replicated_request_done(ar);
#endif
}

static int replmd_replicated_uptodate_search_callback(struct ldb_context *ldb,
						      void *private_data,
						      struct ldb_reply *ares)
{
	struct replmd_replicated_request *ar = talloc_get_type(private_data,
					       struct replmd_replicated_request);
	bool is_done = false;

	switch (ares->type) {
	case LDB_REPLY_ENTRY:
		ar->sub.search_msg = talloc_steal(ar->sub.mem_ctx, ares->message);
		break;
	case LDB_REPLY_REFERRAL:
		/* we ignore referrals */
		break;
	case LDB_REPLY_EXTENDED:
	case LDB_REPLY_DONE:
		is_done = true;
	}

	talloc_free(ares);

#ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */ 
	if (is_done) {
		ar->sub.search_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
		if (ar->sub.search_ret != LDB_SUCCESS) {
			return replmd_replicated_request_error(ar, ar->sub.search_ret);
		}
		if (!ar->sub.search_msg) {
			return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
		}

		return replmd_replicated_uptodate_modify(ar);
	}
#endif
	return LDB_SUCCESS;
}

static int replmd_replicated_uptodate_search(struct replmd_replicated_request *ar)
{
	int ret;
	static const char *attrs[] = {
		"replUpToDateVector",
		"repsFrom",
		NULL
	};

	ret = ldb_build_search_req(&ar->sub.search_req,
				   ar->module->ldb,
				   ar->sub.mem_ctx,
				   ar->objs->partition_dn,
				   LDB_SCOPE_BASE,
				   "(objectClass=*)",
				   attrs,
				   NULL,
				   ar,
				   replmd_replicated_uptodate_search_callback);
	if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);

#ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */ 
	return ldb_next_request(ar->module, ar->sub.search_req);
#else
	ret = ldb_next_request(ar->module, ar->sub.search_req);
	if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);

	ar->sub.search_ret = ldb_wait(ar->sub.search_req->handle, LDB_WAIT_ALL);
	if (ar->sub.search_ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ar->sub.search_ret);
	}
	if (!ar->sub.search_msg) {
		return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
	}

	return replmd_replicated_uptodate_modify(ar);
#endif
}

static int replmd_replicated_uptodate_vector(struct replmd_replicated_request *ar)
{
	ar->sub.mem_ctx = talloc_new(ar);
	if (!ar->sub.mem_ctx) return replmd_replicated_request_werror(ar, WERR_NOMEM);

	return replmd_replicated_uptodate_search(ar);
}

static int replmd_extended_replicated_objects(struct ldb_module *module, struct ldb_request *req)
{
	struct dsdb_extended_replicated_objects *objs;
	struct replmd_replicated_request *ar;

	ldb_debug(module->ldb, LDB_DEBUG_TRACE, "replmd_extended_replicated_objects\n");

	objs = talloc_get_type(req->op.extended.data, struct dsdb_extended_replicated_objects);
	if (!objs) {
		ldb_debug(module->ldb, LDB_DEBUG_FATAL, "replmd_extended_replicated_objects: invalid extended data\n");
		return LDB_ERR_PROTOCOL_ERROR;
	}

	if (objs->version != DSDB_EXTENDED_REPLICATED_OBJECTS_VERSION) {
		ldb_debug(module->ldb, LDB_DEBUG_FATAL, "replmd_extended_replicated_objects: extended data invalid version [%u != %u]\n",
			  objs->version, DSDB_EXTENDED_REPLICATED_OBJECTS_VERSION);
		return LDB_ERR_PROTOCOL_ERROR;
	}

	ar = replmd_replicated_init_handle(module, req, objs);
	if (!ar) {
		return LDB_ERR_OPERATIONS_ERROR;
	}

#ifdef REPLMD_FULL_ASYNC /* TODO: activate this code when ldb support full async code */ 
	return replmd_replicated_apply_next(ar);
#else
	while (ar->index_current < ar->objs->num_objects &&
	       req->handle->state != LDB_ASYNC_DONE) { 
		replmd_replicated_apply_next(ar);
	}

	if (req->handle->state != LDB_ASYNC_DONE) {
		replmd_replicated_uptodate_vector(ar);
	}

	return LDB_SUCCESS;
#endif
}

static int replmd_extended(struct ldb_module *module, struct ldb_request *req)
{
	if (strcmp(req->op.extended.oid, DSDB_EXTENDED_REPLICATED_OBJECTS_OID) == 0) {
		return replmd_extended_replicated_objects(module, req);
	}

	return ldb_next_request(module, req);
}

static int replmd_wait_none(struct ldb_handle *handle) {
	struct replmd_replicated_request *ar;
    
	if (!handle || !handle->private_data) {
		return LDB_ERR_OPERATIONS_ERROR;
	}

	ar = talloc_get_type(handle->private_data, struct replmd_replicated_request);
	if (!ar) {
		return LDB_ERR_OPERATIONS_ERROR;
	}

	/* we do only sync calls */
	if (handle->state != LDB_ASYNC_DONE) {
		return LDB_ERR_OPERATIONS_ERROR;
	}

	return handle->status;
}

static int replmd_wait_all(struct ldb_handle *handle) {

	int ret;

	while (handle->state != LDB_ASYNC_DONE) {
		ret = replmd_wait_none(handle);
		if (ret != LDB_SUCCESS) {
			return ret;
		}
	}

	return handle->status;
}

static int replmd_wait(struct ldb_handle *handle, enum ldb_wait_type type)
{
	if (type == LDB_WAIT_ALL) {
		return replmd_wait_all(handle);
	} else {
		return replmd_wait_none(handle);
	}
}

_PUBLIC_ const struct ldb_module_ops ldb_repl_meta_data_module_ops = {
	.name          = "repl_meta_data",
	.add           = replmd_add,
	.modify        = replmd_modify,
	.extended      = replmd_extended,
	.wait          = replmd_wait
};