/** BEGIN COPYRIGHT BLOCK * 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; version 2 of the License. * * 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, write to the Free Software Foundation, Inc., 59 Temple * Place, Suite 330, Boston, MA 02111-1307 USA. * * In addition, as a special exception, Red Hat, Inc. gives You the additional * right to link the code of this Program with code not covered under the GNU * General Public License ("Non-GPL Code") and to distribute linked combinations * including the two, subject to the limitations in this paragraph. Non-GPL Code * permitted under this exception must only link to the code of this Program * through those well defined interfaces identified in the file named EXCEPTION * found in the source code files (the "Approved Interfaces"). The files of * Non-GPL Code may instantiate templates or use macros or inline functions from * the Approved Interfaces without causing the resulting work to be covered by * the GNU General Public License. Only Red Hat, Inc. may make changes or * additions to the list of Approved Interfaces. You must obey the GNU General * Public License in all respects for all of the Program code and other code used * in conjunction with the Program except the Non-GPL Code covered by this * exception. If you modify this file, you may extend this exception to your * version of the file, but you are not obligated to do so. If you do not wish to * provide this exception without modification, you must delete this exception * statement from your version and license this file solely under the GPL without * exception. * * * Copyright (C) 2001 Sun Microsystems, Inc. Used by permission. * Copyright (C) 2005 Red Hat, Inc. * All rights reserved. * END COPYRIGHT BLOCK **/ #ifdef HAVE_CONFIG_H # include #endif /* repl5_protocol_util.c */ /* Code common to both incremental and total protocols. */ #include "repl5.h" #include "repl5_prot_private.h" /* * Obtain a current CSN (e.g. one that would have been * generated for an operation occurring at this time) * for a given replica. */ CSN * get_current_csn(Slapi_DN *replarea_sdn) { Object *replica_obj; Replica *replica; Object *gen_obj; CSNGen *gen; CSN *current_csn = NULL; if (NULL != replarea_sdn) { replica_obj = replica_get_replica_from_dn(replarea_sdn); if (NULL != replica_obj) { replica = object_get_data(replica_obj); if (NULL != replica) { gen_obj = replica_get_csngen(replica); if (NULL != gen_obj) { gen = (CSNGen *)object_get_data(gen_obj); if (NULL != gen) { if (csngen_new_csn(gen, ¤t_csn, PR_FALSE /* notify */) != CSN_SUCCESS) { csn_free(¤t_csn); } object_release(gen_obj); } } } } } return current_csn; } /* * Acquire exclusive access to a replica. Send a start replication extended * operation to the replica. The response will contain a success code, and * optionally the replica's update vector if acquisition is successful. * This function returns one of the following: * ACQUIRE_SUCCESS - the replica was acquired, and we have exclusive update access * ACQUIRE_REPLICA_BUSY - another master was updating the replica * ACQUIRE_FATAL_ERROR - something bad happened, and it's not likely to improve * if we wait. * ACQUIRE_TRANSIENT_ERROR - something bad happened, but it's probably worth * another try after waiting a while. * If ACQUIRE_SUCCESS is returned, then ruv will point to the replica's update * vector. It's possible that the replica does something goofy and doesn't * return us an update vector, so be prepared for ruv to be NULL (but this is * an error). */ int acquire_replica(Private_Repl_Protocol *prp, char *prot_oid, RUV **ruv) { int return_value; ConnResult crc; Repl_Connection *conn; struct berval *retdata = NULL; char *retoid = NULL; Slapi_DN *replarea_sdn = NULL; struct berval **ruv_bervals = NULL; CSN *current_csn = NULL; PR_ASSERT(prp && prot_oid); if (prp->replica_acquired) /* we already acquire replica */ { slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Remote replica already acquired\n", agmt_get_long_name(prp->agmt)); return_value = ACQUIRE_FATAL_ERROR; return ACQUIRE_SUCCESS; } if (NULL != ruv) { ruv_destroy ( ruv ); } if (strcmp(prot_oid, REPL_NSDS50_INCREMENTAL_PROTOCOL_OID) == 0) { Replica *replica; Object *supl_ruv_obj, *cons_ruv_obj; PRBool is_newer = PR_FALSE; object_acquire(prp->replica_object); replica = object_get_data(prp->replica_object); supl_ruv_obj = replica_get_ruv ( replica ); cons_ruv_obj = agmt_get_consumer_ruv ( prp->agmt ); is_newer = ruv_is_newer ( supl_ruv_obj, cons_ruv_obj ); if ( supl_ruv_obj ) object_release ( supl_ruv_obj ); if ( cons_ruv_obj ) object_release ( cons_ruv_obj ); object_release (prp->replica_object); replica = NULL; if (is_newer == PR_FALSE) { prp->last_acquire_response_code = NSDS50_REPL_UPTODATE; return ACQUIRE_CONSUMER_WAS_UPTODATE; } } prp->last_acquire_response_code = NSDS50_REPL_REPLICA_NO_RESPONSE; /* Get the connection */ conn = prp->conn; crc = conn_connect(conn); if (CONN_OPERATION_FAILED == crc) { return_value = ACQUIRE_TRANSIENT_ERROR; } else if (CONN_SSL_NOT_ENABLED == crc) { return_value = ACQUIRE_FATAL_ERROR; } else { /* we don't want the timer to go off in the middle of an operation */ conn_cancel_linger(conn); /* Does the remote replica support the 5.0 protocol? */ crc = conn_replica_supports_ds5_repl(conn); if (CONN_DOES_NOT_SUPPORT_DS5_REPL == crc) { return_value = ACQUIRE_FATAL_ERROR; } else if (CONN_NOT_CONNECTED == crc || CONN_OPERATION_FAILED == crc) { /* We don't know anything about the remote replica. Try again later. */ return_value = ACQUIRE_TRANSIENT_ERROR; goto error; } /* Find out what level of replication the replica supports. */ crc = conn_replica_supports_ds90_repl(conn); if (CONN_DOES_NOT_SUPPORT_DS90_REPL == crc) { /* Does the remote replica support the 7.1 protocol? */ crc = conn_replica_supports_ds71_repl(conn); if (CONN_DOES_NOT_SUPPORT_DS71_REPL == crc) { /* This is a pre-7.1 replica. */ prp->repl50consumer = 1; } else if (CONN_NOT_CONNECTED == crc || CONN_OPERATION_FAILED == crc) { /* We don't know anything about the remote replica. Try again later. */ return_value = ACQUIRE_TRANSIENT_ERROR; goto error; } else { /* This replica is later than 7.1, but pre-9.0. */ prp->repl71consumer = 1; } } else if (CONN_NOT_CONNECTED == crc || CONN_OPERATION_FAILED == crc) { /* We don't know anything about the remote replica. Try again later. */ return_value = ACQUIRE_TRANSIENT_ERROR; goto error; } else { /* This replica is a 9.0 or later replica. */ prp->repl90consumer = 1; } /* Good to go. Start the protocol. */ /* Obtain a current CSN */ replarea_sdn = agmt_get_replarea(prp->agmt); current_csn = get_current_csn(replarea_sdn); if (NULL != current_csn) { struct berval *payload = NULL; if (prp->repl90consumer) { int is_total = 0; char *data_guid = NULL; struct berval *data = NULL; /* Check if this is a total or incremental update. */ if (strcmp(REPL_NSDS50_TOTAL_PROTOCOL_OID, prot_oid) == 0) { is_total = 1; } /* Call pre-start replication session callback. This callback * may have extra data to be sent to the replica. */ if (repl_session_plugin_call_pre_acquire_cb(prp->agmt, is_total, &data_guid, &data) == 0) { payload = NSDS90StartReplicationRequest_new( prot_oid, slapi_sdn_get_ndn(replarea_sdn), NULL, current_csn, data_guid, data); slapi_ch_free_string(&data_guid); ber_bvfree(data); data = NULL; } else { return_value = ACQUIRE_TRANSIENT_ERROR; slapi_ch_free_string(&data_guid); ber_bvfree(data); data = NULL; goto error; } } else { payload = NSDS50StartReplicationRequest_new( prot_oid, slapi_sdn_get_ndn(replarea_sdn), NULL /* XXXggood need to provide referral(s) */, current_csn); } /* JCMREPL - Need to extract the referrals from the RUV */ crc = conn_send_extended_operation(conn, prp->repl90consumer ? REPL_START_NSDS90_REPLICATION_REQUEST_OID : REPL_START_NSDS50_REPLICATION_REQUEST_OID, payload, NULL /* update control */, NULL /* Message ID */); if (CONN_OPERATION_SUCCESS != crc) { int operation, error; conn_get_error(conn, &operation, &error); /* Couldn't send the extended operation */ return_value = ACQUIRE_TRANSIENT_ERROR; /* XXX right return value? */ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to send a startReplication " "extended operation to consumer (%s). Will retry later.\n", agmt_get_long_name(prp->agmt), error ? ldap_err2string(error) : "unknown error"); } /* Since the operation request is async, we need to wait for the response here */ crc = conn_read_result_ex(conn,&retoid,&retdata,NULL,NULL,1); ber_bvfree(payload); payload = NULL; /* Look at the response we got. */ if (CONN_OPERATION_SUCCESS == crc) { /* * Extop was processed. Look at extop response to see if we're * permitted to go ahead. */ int extop_result; char *data_guid = NULL; struct berval *data = NULL; int extop_rc = decode_repl_ext_response(retdata, &extop_result, &ruv_bervals, &data_guid, &data); if (0 == extop_rc) { prp->last_acquire_response_code = extop_result; switch (extop_result) { /* XXXggood handle other error codes here */ case NSDS50_REPL_INTERNAL_ERROR: slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to acquire replica: " "an internal error occurred on the remote replica. " "Replication is aborting.\n", agmt_get_long_name(prp->agmt)); return_value = ACQUIRE_FATAL_ERROR; break; case NSDS50_REPL_PERMISSION_DENIED: /* Not allowed to send updates */ { char *repl_binddn = agmt_get_binddn(prp->agmt); slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to acquire replica: permission denied. " "The bind dn \"%s\" does not have permission to " "supply replication updates to the replica. " "Will retry later.\n", agmt_get_long_name(prp->agmt), repl_binddn); slapi_ch_free((void **)&repl_binddn); return_value = ACQUIRE_TRANSIENT_ERROR; break; } case NSDS50_REPL_NO_SUCH_REPLICA: /* There is no such replica on the consumer */ { Slapi_DN *repl_root = agmt_get_replarea(prp->agmt); slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to acquire replica: there is no " "replicated area \"%s\" on the consumer server. " "Replication is aborting.\n", agmt_get_long_name(prp->agmt), slapi_sdn_get_dn(repl_root)); slapi_sdn_free(&repl_root); return_value = ACQUIRE_FATAL_ERROR; break; } case NSDS50_REPL_EXCESSIVE_CLOCK_SKEW: /* Large clock skew between the consumer and the supplier */ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to acquire replica: " "Excessive clock skew between the supplier and " "the consumer. Replication is aborting.\n", agmt_get_long_name(prp->agmt)); return_value = ACQUIRE_FATAL_ERROR; break; case NSDS50_REPL_DECODING_ERROR: /* We sent something the replica couldn't understand. */ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to acquire replica: " "the consumer was unable to decode the " "startReplicationRequest extended operation sent by the " "supplier. Replication is aborting.\n", agmt_get_long_name(prp->agmt)); return_value = ACQUIRE_FATAL_ERROR; break; case NSDS50_REPL_REPLICA_BUSY: /* Someone else is updating the replica. Try later. */ /* if acquire_replica is called for replica initialization, log REPLICA_BUSY, too */ if (strcmp(REPL_NSDS50_TOTAL_PROTOCOL_OID, prot_oid) == 0) { slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to acquire replica: " "the replica is currently being updated" "by another supplier.\n", agmt_get_long_name(prp->agmt)); } else /* REPL_NSDS50_INCREMENTAL_PROTOCOL_OID */ { slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, "%s: Unable to acquire replica: " "the replica is currently being updated" "by another supplier. Will try later\n", agmt_get_long_name(prp->agmt)); } return_value = ACQUIRE_REPLICA_BUSY; break; case NSDS50_REPL_LEGACY_CONSUMER: /* remote replica is a legacy consumer */ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to acquire replica: the replica " "is supplied by a legacy supplier. " "Replication is aborting.\n", agmt_get_long_name(prp->agmt)); return_value = ACQUIRE_FATAL_ERROR; break; case NSDS50_REPL_REPLICAID_ERROR: /* remote replica detected a duplicate ReplicaID */ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to aquire replica: the replica " "has the same Replica ID as this one. " "Replication is aborting.\n", agmt_get_long_name(prp->agmt)); return_value = ACQUIRE_FATAL_ERROR; break; case NSDS50_REPL_BACKOFF: /* A replication sesssion hook on the replica * wants us to go into backoff mode. */ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to acquire replica: " "the replica instructed us to go into " "backoff mode. Will retry later.\n", agmt_get_long_name(prp->agmt)); return_value = ACQUIRE_TRANSIENT_ERROR; break; case NSDS50_REPL_REPLICA_READY: /* Call any registered replication session post * acquire callback if we are dealing with a 9.0 * style replica. We want to bail on sending * updates if the return value is non-0. */ if (prp->repl90consumer) { int is_total = 0; /* Check if this is a total or incremental update. */ if (strcmp(REPL_NSDS50_TOTAL_PROTOCOL_OID, prot_oid) == 0) { is_total = 1; } if (repl_session_plugin_call_post_acquire_cb(prp->agmt, is_total, data_guid, data)) { slapi_ch_free_string(&data_guid); ber_bvfree(data); data = NULL; return_value = ACQUIRE_TRANSIENT_ERROR; break; } slapi_ch_free_string(&data_guid); ber_bvfree(data); data = NULL; } /* We've acquired the replica. */ slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, "%s: Replica was successfully acquired.\n", agmt_get_long_name(prp->agmt)); /* Parse the update vector */ if (NULL != ruv_bervals && NULL != ruv) { if (ruv_init_from_bervals(ruv_bervals, ruv) != RUV_SUCCESS) { /* Couldn't parse the update vector */ *ruv = NULL; slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Warning: acquired replica, " "but could not parse update vector. " "The replica must be reinitialized.\n", agmt_get_long_name(prp->agmt)); } } /* Save consumer's RUV in the replication agreement. It is used by the changelog trimming code */ if (ruv && *ruv) agmt_set_consumer_ruv (prp->agmt, *ruv); return_value = ACQUIRE_SUCCESS; break; default: return_value = ACQUIRE_FATAL_ERROR; } } else { /* Couldn't parse the response */ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to parse the response to the " "startReplication extended operation. " "Replication is aborting.\n", agmt_get_long_name(prp->agmt)); prp->last_acquire_response_code = NSDS50_REPL_INTERNAL_ERROR; return_value = ACQUIRE_FATAL_ERROR; } } else { int operation, error; conn_get_error(conn, &operation, &error); /* Couldn't send the extended operation */ return_value = ACQUIRE_TRANSIENT_ERROR; /* XXX right return value? */ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to receive the response for a startReplication " "extended operation to consumer (%s). Will retry later.\n", agmt_get_long_name(prp->agmt), error ? ldap_err2string(error) : "unknown error"); } } else { /* Couldn't get a current CSN */ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to obtain current CSN. " "Replication is aborting.\n", agmt_get_long_name(prp->agmt)); return_value = ACQUIRE_FATAL_ERROR; } } error: csn_free(¤t_csn); if (NULL != ruv_bervals) ber_bvecfree(ruv_bervals); if (NULL != replarea_sdn) slapi_sdn_free(&replarea_sdn); if (NULL != retoid) ldap_memfree(retoid); if (NULL != retdata) ber_bvfree(retdata); if (ACQUIRE_SUCCESS != return_value) { /* could not acquire the replica, so reinstate the linger timer, since this means we won't call release_replica, which also reinstates the timer */ conn_start_linger(conn); } else { /* replica successfully acquired */ prp->replica_acquired = PR_TRUE; } return return_value; } /* * Release a replica by sending an "end replication" extended request. */ void release_replica(Private_Repl_Protocol *prp) { int rc; struct berval *retdata = NULL; char *retoid = NULL; struct berval *payload = NULL; Slapi_DN *replarea_sdn = NULL; int sent_message_id = 0; int ret_message_id = 0; ConnResult conres = 0; PR_ASSERT(NULL != prp); PR_ASSERT(NULL != prp->conn); if (!prp->replica_acquired) return; replarea_sdn = agmt_get_replarea(prp->agmt); payload = NSDS50EndReplicationRequest_new((char *)slapi_sdn_get_dn(replarea_sdn)); /* XXXggood had to cast away const */ slapi_sdn_free(&replarea_sdn); rc = conn_send_extended_operation(prp->conn, REPL_END_NSDS50_REPLICATION_REQUEST_OID, payload, NULL /* update control */, &sent_message_id /* Message ID */); ber_bvfree(payload); /* done with this - free it now */ if (0 != rc) { int operation, error; conn_get_error(prp->conn, &operation, &error); slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Warning: unable to send endReplication extended operation (%s)\n", agmt_get_long_name(prp->agmt), error ? ldap_err2string(error) : "unknown error"); goto error; } /* Since the operation request is async, we need to wait for the response here */ conres = conn_read_result_ex(prp->conn,&retoid,&retdata,NULL,&ret_message_id,1); if (CONN_OPERATION_SUCCESS != conres) { int operation, error; conn_get_error(prp->conn, &operation, &error); slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Warning: unable to receive endReplication extended operation response (%s)\n", agmt_get_long_name(prp->agmt), error ? ldap_err2string(error) : "unknown error"); } else { struct berval **ruv_bervals = NULL; /* Shouldn't actually be returned */ int extop_result; int extop_rc = 0; char *data_guid = NULL; struct berval *data = NULL; /* Check the message id's match */ if (sent_message_id != sent_message_id) { int operation, error; conn_get_error(prp->conn, &operation, &error); slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Warning: response message id does not match the request (%s)\n", agmt_get_long_name(prp->agmt), error ? ldap_err2string(error) : "unknown error"); } /* We need to pass data_guid and data in even though they * are not used here. We will free them anyway in case they * are used in the future. */ extop_rc = decode_repl_ext_response(retdata, &extop_result, (struct berval ***)&ruv_bervals, &data_guid, &data); slapi_ch_free_string(&data_guid); ber_bvfree(data); data = NULL; if (0 == extop_rc) { if (NSDS50_REPL_REPLICA_RELEASE_SUCCEEDED == extop_result) { slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, "%s: Successfully released consumer\n", agmt_get_long_name(prp->agmt)); } else { slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Unable to release consumer: response code %d\n", agmt_get_long_name(prp->agmt), extop_result); /* disconnect from the consumer so that it does not stay locked */ conn_disconnect (prp->conn); } } else { /* Couldn't parse the response */ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "%s: Warning: Unable to parse the response " " to the endReplication extended operation.\n", agmt_get_long_name(prp->agmt)); } if (NULL != ruv_bervals) ber_bvecfree(ruv_bervals); /* XXXggood free ruv_bervals if we got them for some reason */ } if (NULL != retoid) ldap_memfree(retoid); if (NULL != retdata) ber_bvfree(retdata); /* replica is released, start the linger timer on the connection, which was stopped in acquire_replica */ conn_start_linger(prp->conn); error: prp->replica_acquired = PR_FALSE; } /* converts consumer's response to a string */ char * protocol_response2string (int response) { switch (response) { case NSDS50_REPL_REPLICA_READY: return "replica acquired"; case NSDS50_REPL_REPLICA_BUSY: return "replica busy"; case NSDS50_REPL_EXCESSIVE_CLOCK_SKEW: return "excessive clock skew"; case NSDS50_REPL_PERMISSION_DENIED: return "permission denied"; case NSDS50_REPL_DECODING_ERROR: return "decoding error"; case NSDS50_REPL_UNKNOWN_UPDATE_PROTOCOL: return "unknown update protocol"; case NSDS50_REPL_NO_SUCH_REPLICA: return "no such replica"; case NSDS50_REPL_BELOW_PURGEPOINT: return "csn below purge point"; case NSDS50_REPL_INTERNAL_ERROR: return "internal error"; case NSDS50_REPL_REPLICA_RELEASE_SUCCEEDED: return "replica released"; case NSDS50_REPL_LEGACY_CONSUMER: return "replica is a legacy consumer"; case NSDS50_REPL_REPLICAID_ERROR: return "duplicate replica ID detected"; case NSDS50_REPL_UPTODATE: return "no change to send"; default: return "unknown error"; } } int repl5_strip_fractional_mods(Repl_Agmt *agmt, LDAPMod ** mods) { int retval = 0; int i = 0; char **a = agmt_get_fractional_attrs(agmt); if (a) { /* Iterate through the fractional attr list */ for ( i = 0; a[i] != NULL; i++ ) { char *this_excluded_attr = a[i]; int j = 0; for ( j = 0; NULL != mods[ j ]; ) { /* For each one iterate through the attrs in this mod list */ /* For any that match, remove the mod */ LDAPMod *this_mod = mods[j]; if (0 == slapi_attr_type_cmp(this_mod->mod_type,this_excluded_attr,SLAPI_TYPE_CMP_SUBTYPE)) { /* Move down all subsequent mods */ int k = 0; for (k = j; mods[k+1] ; k++) { mods[k] = mods[k+1]; } /* Zero the end of the array */ mods[k] = NULL; /* Adjust value of j, implicit in not incrementing it */ /* Free this mod */ ber_bvecfree(this_mod->mod_bvalues); slapi_ch_free((void **)&(this_mod->mod_type)); slapi_ch_free((void **)&this_mod); } else { j++; } } } slapi_ch_array_free(a); } return retval; }