static int replmd_replicated_apply_merge(struct replmd_replicated_request *ar)
{
	struct ldb_context *ldb;
	struct ldb_request *change_req;
	enum ndr_err_code ndr_err;
	struct ldb_message *msg;
	struct replPropertyMetaDataBlob *rmd;
	struct replPropertyMetaDataBlob omd;
	const struct ldb_val *omd_value;
	struct replPropertyMetaDataBlob nmd;
	struct ldb_val nmd_value;
	struct GUID remote_parent_guid;
	unsigned int i;
	uint32_t j,ni=0;
	unsigned int removed_attrs = 0;
	int ret;
	int (*callback)(struct ldb_request *req, struct ldb_reply *ares) = replmd_op_callback;
	bool isDeleted = false;
	bool local_isDeleted = false;
	bool remote_isDeleted = false;
	bool take_remote_isDeleted = false;
	bool sd_updated = false;
	bool renamed = false;
	bool renamed_to_conflict = false;
	bool is_schema_nc = false;
	NTSTATUS nt_status;
	const struct ldb_val *old_rdn, *new_rdn;
	struct replmd_private *replmd_private =
		talloc_get_type(ldb_module_get_private(ar->module),
				struct replmd_private);
	NTTIME now;
	time_t t = time(NULL);
	unix_to_nt_time(&now, t);

	ldb = ldb_module_get_ctx(ar->module);
	msg = ar->objs->objects[ar->index_current].msg;

	is_schema_nc = ldb_dn_compare_base(replmd_private->schema_dn, msg->dn) == 0;

	rmd = ar->objs->objects[ar->index_current].meta_data;
	ZERO_STRUCT(omd);
	omd.version = 1;

	/* find existing meta data */
	omd_value = ldb_msg_find_ldb_val(ar->search_msg, "replPropertyMetaData");
	if (omd_value) {
		ndr_err = ndr_pull_struct_blob(omd_value, ar, &omd,
					       (ndr_pull_flags_fn_t)ndr_pull_replPropertyMetaDataBlob);
		if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
			nt_status = ndr_map_error2ntstatus(ndr_err);
			return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
		}

		if (omd.version != 1) {
			return replmd_replicated_request_werror(ar, WERR_DS_DRA_INTERNAL_ERROR);
		}
	}

	if (DEBUGLVL(8)) {
		struct GUID_txt_buf guid_txt;

		char *s = ldb_ldif_message_redacted_string(ldb, ar,
							   LDB_CHANGETYPE_MODIFY, msg);
		DEBUG(8, ("Initial DRS replication modify message of %s is:\n%s\n"
			  "%s\n"
			  "%s\n",
			  GUID_buf_string(&ar->objs->objects[ar->index_current].object_guid, &guid_txt),
			  s,
			  ndr_print_struct_string(s,
						  (ndr_print_fn_t)ndr_print_replPropertyMetaDataBlob,
						  "existing replPropertyMetaData",
						  &omd),
			  ndr_print_struct_string(s,
						  (ndr_print_fn_t)ndr_print_replPropertyMetaDataBlob,
						  "incoming replPropertyMetaData",
						  rmd)));
		talloc_free(s);
	} else if (DEBUGLVL(4)) {
		struct GUID_txt_buf guid_txt;

		DEBUG(4, ("Initial DRS replication modify DN of %s is: %s\n",
			  GUID_buf_string(&ar->objs->objects[ar->index_current].object_guid,
					  &guid_txt),
			  ldb_dn_get_linearized(msg->dn)));
	}
		
	local_isDeleted = ldb_msg_find_attr_as_bool(ar->search_msg,
						    "isDeleted", false);
	remote_isDeleted = ldb_msg_find_attr_as_bool(msg,
						     "isDeleted", false);

	/*
	 * Fill in the remote_parent_guid with the GUID or an all-zero
	 * GUID.
	 */
	if (ar->objs->objects[ar->index_current].parent_guid != NULL) {
		remote_parent_guid = *ar->objs->objects[ar->index_current].parent_guid;
	} else {
		remote_parent_guid = GUID_zero();
	}

	/*
	 * To ensure we follow a complex rename chain around, we have
	 * to confirm that the DN is the same (mostly to confirm the
	 * RDN) and the parentGUID is the same.
	 *
	 * This ensures we keep things under the correct parent, which
	 * replmd_replicated_handle_rename() will do.
	 */

	if (strcmp(ldb_dn_get_linearized(msg->dn), ldb_dn_get_linearized(ar->search_msg->dn)) == 0
	    && GUID_equal(&remote_parent_guid, &ar->local_parent_guid)) {
		ret = LDB_SUCCESS;
	} else {
		/*
		 * handle renames, even just by case that come in over
		 * DRS.  Changes in the parent DN don't hit us here,
		 * because the search for a parent will clean up those
		 * components.
		 *
		 * We also have already filtered out the case where
		 * the peer has an older name to what we have (see
		 * replmd_replicated_apply_search_callback())
		 */
		ret = replmd_replicated_handle_rename(ar, msg, ar->req, &renamed_to_conflict);

		/*
		 * This looks strange, but we must set this after any
		 * rename, otherwise the SD propegation will not
		 * happen (which might matter if we have a new parent)
		 *
		 * The additional case of calling
		 * replmd_op_name_modify_callback (below) is
		 * controlled by renamed_to_conflict.
		 */
		renamed = true;
	}

	if (ret != LDB_SUCCESS) {
		ldb_debug(ldb, LDB_DEBUG_FATAL,
			  "replmd_replicated_request rename %s => %s failed - %s\n",
			  ldb_dn_get_linearized(ar->search_msg->dn),
			  ldb_dn_get_linearized(msg->dn),
			  ldb_errstring(ldb));
		return replmd_replicated_request_werror(ar, WERR_DS_DRA_DB_ERROR);
	}

	if (renamed_to_conflict == true) {
		/*
		 * Set the callback to one that will fix up the name
		 * metadata on the new conflict DN
		 */
		callback = replmd_op_name_modify_callback;
	}

	ZERO_STRUCT(nmd);
	nmd.version = 1;
	nmd.ctr.ctr1.count = omd.ctr.ctr1.count + rmd->ctr.ctr1.count;
	nmd.ctr.ctr1.array = talloc_array(ar,
					  struct replPropertyMetaData1,
					  nmd.ctr.ctr1.count);
	if (!nmd.ctr.ctr1.array) return replmd_replicated_request_werror(ar, WERR_NOT_ENOUGH_MEMORY);

	/* first copy the old meta data */
	for (i=0; i < omd.ctr.ctr1.count; i++) {
		nmd.ctr.ctr1.array[ni]	= omd.ctr.ctr1.array[i];
		ni++;
	}

	ar->seq_num = 0;
	/* now merge in the new meta data */
	for (i=0; i < rmd->ctr.ctr1.count; i++) {
		bool found = false;

		for (j=0; j < ni; j++) {
			bool cmp;

			if (rmd->ctr.ctr1.array[i].attid != nmd.ctr.ctr1.array[j].attid) {
				continue;
			}

			cmp = replmd_replPropertyMetaData1_new_should_be_taken(
				ar->objs->dsdb_repl_flags,
				&nmd.ctr.ctr1.array[j],
				&rmd->ctr.ctr1.array[i]);
			if (cmp) {
				/* replace the entry */
				nmd.ctr.ctr1.array[j] = rmd->ctr.ctr1.array[i];
				if (ar->seq_num == 0) {
					ret = ldb_sequence_number(ldb, LDB_SEQ_NEXT, &ar->seq_num);
					if (ret != LDB_SUCCESS) {
						return replmd_replicated_request_error(ar, ret);
					}
				}
				nmd.ctr.ctr1.array[j].local_usn = ar->seq_num;
				switch (nmd.ctr.ctr1.array[j].attid) {
				case DRSUAPI_ATTID_ntSecurityDescriptor:
					sd_updated = true;
					break;
				case DRSUAPI_ATTID_isDeleted:
					take_remote_isDeleted = true;
					break;
				default:
					break;
				}
				found = true;
				break;
			}

			if (rmd->ctr.ctr1.array[i].attid != DRSUAPI_ATTID_instanceType) {
				DEBUG(3,("Discarding older DRS attribute update to %s on %s from %s\n",
					 msg->elements[i-removed_attrs].name,
					 ldb_dn_get_linearized(msg->dn),
					 GUID_string(ar, &rmd->ctr.ctr1.array[i].originating_invocation_id)));
			}

			/* we don't want to apply this change so remove the attribute */
			ldb_msg_remove_element(msg, &msg->elements[i-removed_attrs]);
			removed_attrs++;

			found = true;
			break;
		}

		if (found) continue;

		nmd.ctr.ctr1.array[ni] = rmd->ctr.ctr1.array[i];
		if (ar->seq_num == 0) {
			ret = ldb_sequence_number(ldb, LDB_SEQ_NEXT, &ar->seq_num);
			if (ret != LDB_SUCCESS) {
				return replmd_replicated_request_error(ar, ret);
			}
		}
		nmd.ctr.ctr1.array[ni].local_usn = ar->seq_num;
		switch (nmd.ctr.ctr1.array[ni].attid) {
		case DRSUAPI_ATTID_ntSecurityDescriptor:
			sd_updated = true;
			break;
		case DRSUAPI_ATTID_isDeleted:
			take_remote_isDeleted = true;
			break;
		default:
			break;
		}
		ni++;
	}

	/*
	 * finally correct the size of the meta_data array
	 */
	nmd.ctr.ctr1.count = ni;

	new_rdn = ldb_dn_get_rdn_val(msg->dn);
	old_rdn = ldb_dn_get_rdn_val(ar->search_msg->dn);

	if (renamed) {
		ret = replmd_update_rpmd_rdn_attr(ldb, msg, new_rdn, old_rdn,
						  &nmd, ar, now, is_schema_nc,
						  false);
		if (ret != LDB_SUCCESS) {
			ldb_asprintf_errstring(ldb, "%s: error during DRS repl merge: %s", __func__, ldb_errstring(ldb));
			return replmd_replicated_request_error(ar, ret);
		}
	}
	/*
	 * sort the new meta data array
	 */
	ret = replmd_replPropertyMetaDataCtr1_sort_and_verify(ldb, &nmd.ctr.ctr1, msg->dn);
	if (ret != LDB_SUCCESS) {
		ldb_asprintf_errstring(ldb, "%s: error during DRS repl merge: %s", __func__, ldb_errstring(ldb));
		return ret;
	}

	/*
	 * Work out if this object is deleted, so we can prune any extra attributes.  See MS-DRSR 4.1.10.6.9
	 * UpdateObject.
	 *
	 * This also controls SD propagation below
	 */
	if (take_remote_isDeleted) {
		isDeleted = remote_isDeleted;
	} else {
		isDeleted = local_isDeleted;
	}

	ar->isDeleted = isDeleted;

	/*
	 * check if some replicated attributes left, otherwise skip the ldb_modify() call
	 */
	if (msg->num_elements == 0) {
		ldb_debug(ldb, LDB_DEBUG_TRACE, "replmd_replicated_apply_merge[%u]: skip replace\n",
			  ar->index_current);

		return replmd_replicated_apply_isDeleted(ar);
	}

	ldb_debug(ldb, LDB_DEBUG_TRACE, "replmd_replicated_apply_merge[%u]: replace %u attributes\n",
		  ar->index_current, msg->num_elements);

	if (renamed) {
		/*
		 * This is an new name for this object, so we must
		 * inherit from the parent
		 *
		 * This is needed because descriptor is above
		 * repl_meta_data in the module stack, so this will
		 * not be trigered 'naturally' by the flow of
		 * operations.
		 */
		ret = dsdb_module_schedule_sd_propagation(ar->module,
							  ar->objs->partition_dn,
							  ar->objs->objects[ar->index_current].object_guid,
							  ar->objs->objects[ar->index_current].parent_guid ?
							  *ar->objs->objects[ar->index_current].parent_guid :
							  GUID_zero(),
							  true);
		if (ret != LDB_SUCCESS) {
			return ldb_operr(ldb);
		}
	}

	if (sd_updated && !isDeleted) {
		/*
		 * This is an existing object, so there is no need to
		 * inherit from the parent, but we must inherit any
		 * incoming changes to our child objects.
		 *
		 * This is needed because descriptor is above
		 * repl_meta_data in the module stack, so this will
		 * not be trigered 'naturally' by the flow of
		 * operations.
		 */
		ret = dsdb_module_schedule_sd_propagation(ar->module,
							  ar->objs->partition_dn,
							  ar->objs->objects[ar->index_current].object_guid,
							  ar->objs->objects[ar->index_current].parent_guid ?
							  *ar->objs->objects[ar->index_current].parent_guid :
							  GUID_zero(),
							  false);
		if (ret != LDB_SUCCESS) {
			return ldb_operr(ldb);
		}
	}

	/* create the meta data value */
	ndr_err = ndr_push_struct_blob(&nmd_value, msg, &nmd,
				       (ndr_push_flags_fn_t)ndr_push_replPropertyMetaDataBlob);
	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
		nt_status = ndr_map_error2ntstatus(ndr_err);
		return replmd_replicated_request_werror(ar, ntstatus_to_werror(nt_status));
	}

	/*
	 * when we know that we'll modify the record, add the whenChanged, uSNChanged
	 * and replPopertyMetaData attributes
	 */
	ret = ldb_msg_add_string(msg, "whenChanged", ar->objs->objects[ar->index_current].when_changed);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}
	ret = samdb_msg_add_uint64(ldb, msg, msg, "uSNChanged", ar->seq_num);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}
	ret = ldb_msg_add_value(msg, "replPropertyMetaData", &nmd_value, NULL);
	if (ret != LDB_SUCCESS) {
		return replmd_replicated_request_error(ar, ret);
	}

	replmd_ldb_message_sort(msg, ar->schema);

	/* we want to replace the old values */
	for (i=0; i < msg->num_elements; i++) {
		msg->elements[i].flags = LDB_FLAG_MOD_REPLACE;
		if (ldb_attr_cmp(msg->elements[i].name, "objectClass") == 0) {
			if (msg->elements[i].num_values == 0) {
				ldb_asprintf_errstring(ldb, __location__
						       ": objectClass removed on %s, aborting replication\n",
						       ldb_dn_get_linearized(msg->dn));
				return replmd_replicated_request_error(ar, LDB_ERR_OBJECT_CLASS_VIOLATION);
			}
		}
	}

	if (DEBUGLVL(8)) {
		struct GUID_txt_buf guid_txt;

		char *s = ldb_ldif_message_redacted_string(ldb, ar,
							   LDB_CHANGETYPE_MODIFY,
							   msg);
		DEBUG(8, ("Final DRS replication modify message of %s:\n%s\n",
			  GUID_buf_string(&ar->objs->objects[ar->index_current].object_guid,
					  &guid_txt),
			  s));
		talloc_free(s);
	} else if (DEBUGLVL(4)) {
		struct GUID_txt_buf guid_txt;

		DEBUG(4, ("Final DRS replication modify DN of %s is %s\n",
			  GUID_buf_string(&ar->objs->objects[ar->index_current].object_guid,
					  &guid_txt),
			  ldb_dn_get_linearized(msg->dn)));
	}

	ret = ldb_build_mod_req(&change_req,
				ldb,
				ar,
				msg,
				ar->controls,
				ar,
				callback,
				ar->req);
	LDB_REQ_SET_LOCATION(change_req);
	if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);

	/* current partition control needed by "repmd_op_callback" */
	ret = ldb_request_add_control(change_req,
				      DSDB_CONTROL_CURRENT_PARTITION_OID,
				      false, NULL);
	if (ret != LDB_SUCCESS) return replmd_replicated_request_error(ar, ret);

	return ldb_next_request(ar->module, change_req);
}