No OneTemporary
Actions

Size

181 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/daemons/based/based_callbacks.c b/daemons/based/based_callbacks.c
	index 382a94f888..8c7e95f7de 100644
	--- a/daemons/based/based_callbacks.c
	+++ b/daemons/based/based_callbacks.c
	@@ -1,1878 +1,1887 @@
	/*
	* Copyright 2004-2023 the Pacemaker project contributors
	*
	* The version control history for this file may have further details.
	*
	* This source code is licensed under the GNU General Public License version 2
	* or later (GPLv2+) WITHOUT ANY WARRANTY.
	*/

	#include <crm_internal.h>

	#include <sys/param.h>
	#include <stdio.h>
	#include <sys/types.h>
	#include <unistd.h>

	#include <stdlib.h>
	#include <stdint.h> // uint32_t, uint64_t, UINT64_C()
	#include <errno.h>
	#include <fcntl.h>
	#include <inttypes.h> // PRIu64

	#include <glib.h>
	#include <libxml/tree.h>

	#include <crm/crm.h>
	#include <crm/cib.h>
	#include <crm/msg_xml.h>
	#include <crm/cluster/internal.h>

	#include <crm/common/xml.h>
	#include <crm/common/remote_internal.h>

	#include <pacemaker-based.h>

	#define EXIT_ESCALATION_MS 10000

	static unsigned long cib_local_bcast_num = 0;

	typedef struct cib_local_notify_s {
	xmlNode *notify_src;
	char *client_id;
	gboolean from_peer;
	gboolean sync_reply;
	} cib_local_notify_t;

	struct digest_data {
	char *nodes;
	char *alerts;
	char *status;
	};

	int next_client_id = 0;

	gboolean legacy_mode = FALSE;

	qb_ipcs_service_t *ipcs_ro = NULL;
	qb_ipcs_service_t *ipcs_rw = NULL;
	qb_ipcs_service_t *ipcs_shm = NULL;

	static int cib_process_command(xmlNode *request,
	const cib_operation_t *operation,
	- xmlNode reply, xmlNode cib_diff,
	- bool privileged);
	+ cib__op_fn_t op_function, xmlNode **reply,
	+ xmlNode **cib_diff, bool privileged);

	static gboolean cib_common_callback(qb_ipcs_connection_t c, void data,
	size_t size, gboolean privileged);

	gboolean
	cib_legacy_mode(void)
	{
	return legacy_mode;
	}

	/*!
	* \internal
	* \brief Free a <tt>struct digest_data</tt> object's members
	*
	* \param[in,out] digests Object whose members to free
	*
	* \note This does not free the object itself, which may be stack-allocated.
	*/
	static void
	free_digests(struct digest_data *digests)
	{
	if (digests != NULL) {
	free(digests->nodes);
	free(digests->alerts);
	free(digests->status);
	}
	}

	static int32_t
	cib_ipc_accept(qb_ipcs_connection_t * c, uid_t uid, gid_t gid)
	{
	if (cib_shutdown_flag) {
	crm_info("Ignoring new IPC client [%d] during shutdown",
	pcmk__client_pid(c));
	return -EPERM;
	}

	if (pcmk__new_client(c, uid, gid) == NULL) {
	return -EIO;
	}
	return 0;
	}

	static int32_t
	cib_ipc_dispatch_rw(qb_ipcs_connection_t * c, void *data, size_t size)
	{
	pcmk__client_t *client = pcmk__find_client(c);

	crm_trace("%p message from %s", c, client->id);
	return cib_common_callback(c, data, size, TRUE);
	}

	static int32_t
	cib_ipc_dispatch_ro(qb_ipcs_connection_t * c, void *data, size_t size)
	{
	pcmk__client_t *client = pcmk__find_client(c);

	crm_trace("%p message from %s", c, client->id);
	return cib_common_callback(c, data, size, FALSE);
	}

	/* Error code means? */
	static int32_t
	cib_ipc_closed(qb_ipcs_connection_t * c)
	{
	pcmk__client_t *client = pcmk__find_client(c);

	if (client == NULL) {
	return 0;
	}
	crm_trace("Connection %p", c);
	based_discard_transaction(client);
	pcmk__free_client(client);
	return 0;
	}

	static void
	cib_ipc_destroy(qb_ipcs_connection_t * c)
	{
	crm_trace("Connection %p", c);
	cib_ipc_closed(c);
	if (cib_shutdown_flag) {
	cib_shutdown(0);
	}
	}

	struct qb_ipcs_service_handlers ipc_ro_callbacks = {
	.connection_accept = cib_ipc_accept,
	.connection_created = NULL,
	.msg_process = cib_ipc_dispatch_ro,
	.connection_closed = cib_ipc_closed,
	.connection_destroyed = cib_ipc_destroy
	};

	struct qb_ipcs_service_handlers ipc_rw_callbacks = {
	.connection_accept = cib_ipc_accept,
	.connection_created = NULL,
	.msg_process = cib_ipc_dispatch_rw,
	.connection_closed = cib_ipc_closed,
	.connection_destroyed = cib_ipc_destroy
	};

	/*!
	* \internal
	* \brief Create reply XML for a CIB request
	*
	* \param[in] op CIB operation type
	* \param[in] call_id CIB call ID
	* \param[in] client_id CIB client ID
	* \param[in] call_options Group of <tt>enum cib_call_options</tt> flags
	* \param[in] rc Request return code
	* \param[in] call_data Request output data
	*
	* \return Reply XML
	*
	* \note The caller is responsible for freeing the return value using
	* \p free_xml().
	*/
	static xmlNode *
	create_cib_reply(const char op, const char call_id, const char *client_id,
	int call_options, int rc, xmlNode *call_data)
	{
	xmlNode *reply = create_xml_node(NULL, "cib-reply");

	CRM_ASSERT(reply != NULL);

	crm_xml_add(reply, F_TYPE, T_CIB);
	crm_xml_add(reply, F_CIB_OPERATION, op);
	crm_xml_add(reply, F_CIB_CALLID, call_id);
	crm_xml_add(reply, F_CIB_CLIENTID, client_id);
	crm_xml_add_int(reply, F_CIB_CALLOPTS, call_options);
	crm_xml_add_int(reply, F_CIB_RC, rc);

	if (call_data != NULL) {
	crm_trace("Attaching reply output");
	add_message_xml(reply, F_CIB_CALLDATA, call_data);
	}

	crm_log_xml_explicit(reply, "cib:reply");
	return reply;
	}

	static void
	do_local_notify(xmlNode notify_src, const char client_id, bool sync_reply,
	bool from_peer)
	{
	int rid = 0;
	int call_id = 0;
	pcmk__client_t *client_obj = NULL;

	CRM_ASSERT(notify_src && client_id);

	crm_element_value_int(notify_src, F_CIB_CALLID, &call_id);

	client_obj = pcmk__find_client_by_id(client_id);
	if (client_obj == NULL) {
	crm_debug("Could not send response %d: client %s not found",
	call_id, client_id);
	return;
	}

	if (sync_reply) {
	if (client_obj->ipcs) {
	CRM_LOG_ASSERT(client_obj->request_id);

	rid = client_obj->request_id;
	client_obj->request_id = 0;

	crm_trace("Sending response %d to client %s%s",
	rid, pcmk__client_name(client_obj),
	(from_peer? " (originator of delegated request)" : ""));
	} else {
	crm_trace("Sending response (call %d) to client %s%s",
	call_id, pcmk__client_name(client_obj),
	(from_peer? " (originator of delegated request)" : ""));
	}

	} else {
	crm_trace("Sending event %d to client %s%s",
	call_id, pcmk__client_name(client_obj),
	(from_peer? " (originator of delegated request)" : ""));
	}

	switch (PCMK__CLIENT_TYPE(client_obj)) {
	case pcmk__client_ipc:
	{
	int rc = pcmk__ipc_send_xml(client_obj, rid, notify_src,
	(sync_reply? crm_ipc_flags_none
	: crm_ipc_server_event));

	if (rc != pcmk_rc_ok) {
	crm_warn("%s reply to client %s failed: %s " CRM_XS " rc=%d",
	(sync_reply? "Synchronous" : "Asynchronous"),
	pcmk__client_name(client_obj), pcmk_rc_str(rc),
	rc);
	}
	}
	break;
	#ifdef HAVE_GNUTLS_GNUTLS_H
	case pcmk__client_tls:
	#endif
	case pcmk__client_tcp:
	pcmk__remote_send_xml(client_obj->remote, notify_src);
	break;
	default:
	crm_err("Unknown transport for client %s "
	CRM_XS " flags=%#016" PRIx64,
	pcmk__client_name(client_obj), client_obj->flags);
	}
	}

	void
	cib_common_callback_worker(uint32_t id, uint32_t flags, xmlNode * op_request,
	pcmk__client_t *cib_client, gboolean privileged)
	{
	const char *op = crm_element_value(op_request, F_CIB_OPERATION);
	int call_options = cib_none;

	crm_element_value_int(op_request, F_CIB_CALLOPTS, &call_options);

	if (pcmk_is_set(call_options, cib_transaction)) {
	// Append request to client's transaction
	const char *call_id = crm_element_value(op_request, F_CIB_CALLID);

	int rc = based_extend_transaction(cib_client, op_request, privileged);

	xmlNode *reply = create_cib_reply(op, call_id, cib_client->id,
	call_options, rc, NULL);

	if (rc != pcmk_rc_ok) {
	crm_warn("Could not extend transaction for client %s: %s",
	pcmk__client_name(cib_client), pcmk_rc_str(rc));
	}

	do_local_notify(reply, cib_client->id,
	pcmk_is_set(call_options, cib_sync_call), false);
	return;
	}

	if (pcmk__str_eq(op, CRM_OP_REGISTER, pcmk__str_none)) {
	if (flags & crm_ipc_client_response) {
	xmlNode *ack = create_xml_node(NULL, __func__);

	crm_xml_add(ack, F_CIB_OPERATION, CRM_OP_REGISTER);
	crm_xml_add(ack, F_CIB_CLIENTID, cib_client->id);
	pcmk__ipc_send_xml(cib_client, id, ack, flags);
	cib_client->request_id = 0;
	free_xml(ack);
	}
	return;

	} else if (pcmk__str_eq(op, T_CIB_NOTIFY, pcmk__str_none)) {
	/* Update the notify filters for this client */
	int on_off = 0;
	crm_exit_t status = CRM_EX_OK;
	uint64_t bit = UINT64_C(0);
	const char *type = crm_element_value(op_request, F_CIB_NOTIFY_TYPE);

	crm_element_value_int(op_request, F_CIB_NOTIFY_ACTIVATE, &on_off);

	crm_debug("Setting %s callbacks %s for client %s",
	type, (on_off? "on" : "off"), pcmk__client_name(cib_client));

	if (pcmk__str_eq(type, T_CIB_POST_NOTIFY, pcmk__str_casei)) {
	bit = cib_notify_post;

	} else if (pcmk__str_eq(type, T_CIB_PRE_NOTIFY, pcmk__str_casei)) {
	bit = cib_notify_pre;

	} else if (pcmk__str_eq(type, T_CIB_UPDATE_CONFIRM, pcmk__str_casei)) {
	bit = cib_notify_confirm;

	} else if (pcmk__str_eq(type, T_CIB_DIFF_NOTIFY, pcmk__str_casei)) {
	bit = cib_notify_diff;

	} else if (pcmk__str_eq(type, T_CIB_REPLACE_NOTIFY, pcmk__str_casei)) {
	bit = cib_notify_replace;

	} else {
	status = CRM_EX_INVALID_PARAM;
	}

	if (bit != 0) {
	if (on_off) {
	pcmk__set_client_flags(cib_client, bit);
	} else {
	pcmk__clear_client_flags(cib_client, bit);
	}
	}

	pcmk__ipc_send_ack(cib_client, id, flags, "ack", NULL, status);
	return;
	}

	cib_process_request(op_request, privileged, cib_client);
	}

	int32_t
	cib_common_callback(qb_ipcs_connection_t * c, void *data, size_t size, gboolean privileged)
	{
	uint32_t id = 0;
	uint32_t flags = 0;
	int call_options = 0;
	pcmk__client_t *cib_client = pcmk__find_client(c);
	xmlNode *op_request = pcmk__client_data2xml(cib_client, data, &id, &flags);

	if (op_request) {
	crm_element_value_int(op_request, F_CIB_CALLOPTS, &call_options);
	}

	if (op_request == NULL) {
	crm_trace("Invalid message from %p", c);
	pcmk__ipc_send_ack(cib_client, id, flags, "nack", NULL, CRM_EX_PROTOCOL);
	return 0;

	} else if(cib_client == NULL) {
	crm_trace("Invalid client %p", c);
	return 0;
	}

	if (pcmk_is_set(call_options, cib_sync_call)) {
	CRM_LOG_ASSERT(flags & crm_ipc_client_response);
	CRM_LOG_ASSERT(cib_client->request_id == 0); /* This means the client has two synchronous events in-flight */
	cib_client->request_id = id; /* Reply only to the last one */
	}

	if (cib_client->name == NULL) {
	const char *value = crm_element_value(op_request, F_CIB_CLIENTNAME);

	if (value == NULL) {
	cib_client->name = pcmk__itoa(cib_client->pid);
	} else {
	cib_client->name = strdup(value);
	if (crm_is_daemon_name(value)) {
	pcmk__set_client_flags(cib_client, cib_is_daemon);
	}
	}
	}

	/* Allow cluster daemons more leeway before being evicted */
	if (pcmk_is_set(cib_client->flags, cib_is_daemon)) {
	const char *qmax = cib_config_lookup("cluster-ipc-limit");

	if (pcmk__set_client_queue_max(cib_client, qmax)) {
	crm_trace("IPC threshold for client %s[%u] is now %u",
	pcmk__client_name(cib_client), cib_client->pid,
	cib_client->queue_max);
	}
	}

	crm_xml_add(op_request, F_CIB_CLIENTID, cib_client->id);
	crm_xml_add(op_request, F_CIB_CLIENTNAME, cib_client->name);

	CRM_LOG_ASSERT(cib_client->user != NULL);
	pcmk__update_acl_user(op_request, F_CIB_USER, cib_client->user);

	cib_common_callback_worker(id, flags, op_request, cib_client, privileged);
	free_xml(op_request);

	return 0;
	}

	static uint64_t ping_seq = 0;
	static char *ping_digest = NULL;
	static bool ping_modified_since = FALSE;

	static gboolean
	cib_digester_cb(gpointer data)
	{
	if (based_is_primary) {
	char buffer[32];
	xmlNode *ping = create_xml_node(NULL, "ping");

	ping_seq++;
	free(ping_digest);
	ping_digest = NULL;
	ping_modified_since = FALSE;
	snprintf(buffer, 32, "%" PRIu64, ping_seq);
	crm_trace("Requesting peer digests (%s)", buffer);

	crm_xml_add(ping, F_TYPE, "cib");
	crm_xml_add(ping, F_CIB_OPERATION, CRM_OP_PING);
	crm_xml_add(ping, F_CIB_PING_ID, buffer);

	crm_xml_add(ping, XML_ATTR_CRM_VERSION, CRM_FEATURE_SET);
	send_cluster_message(NULL, crm_msg_cib, ping, TRUE);

	free_xml(ping);
	}
	return FALSE;
	}

	static void
	process_ping_reply(xmlNode *reply)
	{
	uint64_t seq = 0;
	const char *host = crm_element_value(reply, F_ORIG);

	xmlNode *pong = get_message_xml(reply, F_CIB_CALLDATA);
	const char *seq_s = crm_element_value(pong, F_CIB_PING_ID);
	const char *digest = crm_element_value(pong, XML_ATTR_DIGEST);

	if (seq_s == NULL) {
	crm_debug("Ignoring ping reply with no " F_CIB_PING_ID);
	return;

	} else {
	long long seq_ll;

	if (pcmk__scan_ll(seq_s, &seq_ll, 0LL) != pcmk_rc_ok) {
	return;
	}
	seq = (uint64_t) seq_ll;
	}

	if(digest == NULL) {
	crm_trace("Ignoring ping reply %s from %s with no digest", seq_s, host);

	} else if(seq != ping_seq) {
	crm_trace("Ignoring out of sequence ping reply %s from %s", seq_s, host);

	} else if(ping_modified_since) {
	crm_trace("Ignoring ping reply %s from %s: cib updated since", seq_s, host);

	} else {
	const char *version = crm_element_value(pong, XML_ATTR_CRM_VERSION);

	if(ping_digest == NULL) {
	crm_trace("Calculating new digest");
	ping_digest = calculate_xml_versioned_digest(the_cib, FALSE, TRUE, version);
	}

	crm_trace("Processing ping reply %s from %s (%s)", seq_s, host, digest);
	if (!pcmk__str_eq(ping_digest, digest, pcmk__str_casei)) {
	xmlNode *remote_cib = get_message_xml(pong, F_CIB_CALLDATA);

	crm_notice("Local CIB %s.%s.%s.%s differs from %s: %s.%s.%s.%s %p",
	crm_element_value(the_cib, XML_ATTR_GENERATION_ADMIN),
	crm_element_value(the_cib, XML_ATTR_GENERATION),
	crm_element_value(the_cib, XML_ATTR_NUMUPDATES),
	ping_digest, host,
	remote_cib?crm_element_value(remote_cib, XML_ATTR_GENERATION_ADMIN):"_",
	remote_cib?crm_element_value(remote_cib, XML_ATTR_GENERATION):"_",
	remote_cib?crm_element_value(remote_cib, XML_ATTR_NUMUPDATES):"_",
	digest, remote_cib);

	if(remote_cib && remote_cib->children) {
	// Additional debug
	xml_calculate_changes(the_cib, remote_cib);

	pcmk__output_set_log_level(logger_out, LOG_INFO);
	pcmk__xml_show_changes(logger_out, remote_cib);
	crm_trace("End of differences");
	}

	free_xml(remote_cib);
	sync_our_cib(reply, FALSE);
	}
	}
	}

	static void
	local_notify_destroy_callback(gpointer data)
	{
	cib_local_notify_t *notify = data;

	free_xml(notify->notify_src);
	free(notify->client_id);
	free(notify);
	}

	static void
	check_local_notify(int bcast_id)
	{
	cib_local_notify_t *notify = NULL;

	if (!local_notify_queue) {
	return;
	}

	notify = pcmk__intkey_table_lookup(local_notify_queue, bcast_id);

	if (notify) {
	do_local_notify(notify->notify_src, notify->client_id, notify->sync_reply,
	notify->from_peer);
	pcmk__intkey_table_remove(local_notify_queue, bcast_id);
	}
	}

	static void
	queue_local_notify(xmlNode * notify_src, const char *client_id, gboolean sync_reply,
	gboolean from_peer)
	{
	cib_local_notify_t *notify = calloc(1, sizeof(cib_local_notify_t));

	notify->notify_src = notify_src;
	notify->client_id = strdup(client_id);
	notify->sync_reply = sync_reply;
	notify->from_peer = from_peer;

	if (!local_notify_queue) {
	local_notify_queue = pcmk__intkey_table(local_notify_destroy_callback);
	}
	pcmk__intkey_table_insert(local_notify_queue, cib_local_bcast_num, notify);
	// cppcheck doesn't know notify will get freed when hash table is destroyed
	// cppcheck-suppress memleak
	}

	static void
	parse_local_options_v1(const pcmk__client_t *cib_client,
	const cib_operation_t *operation, int call_options,
	const char host, const char op, gboolean *local_notify,
	gboolean needs_reply, gboolean process,
	gboolean *needs_forward)
	{
	if (pcmk_is_set(operation->flags, cib_op_attr_modifies)
	&& !pcmk_is_set(call_options, cib_inhibit_bcast)) {
	/* we need to send an update anyway */
	*needs_reply = TRUE;
	} else {
	*needs_reply = FALSE;
	}

	if (host == NULL && (call_options & cib_scope_local)) {
	crm_trace("Processing locally scoped %s op from client %s",
	op, pcmk__client_name(cib_client));
	*local_notify = TRUE;

	} else if ((host == NULL) && based_is_primary) {
	crm_trace("Processing %s op locally from client %s as primary",
	op, pcmk__client_name(cib_client));
	*local_notify = TRUE;

	} else if (pcmk__str_eq(host, OUR_NODENAME, pcmk__str_casei)) {
	crm_trace("Processing locally addressed %s op from client %s",
	op, pcmk__client_name(cib_client));
	*local_notify = TRUE;

	} else if (stand_alone) {
	*needs_forward = FALSE;
	*local_notify = TRUE;
	*process = TRUE;

	} else {
	crm_trace("%s op from %s needs to be forwarded to client %s",
	op, pcmk__client_name(cib_client),
	pcmk__s(host, "the primary instance"));
	*needs_forward = TRUE;
	*process = FALSE;
	}
	}

	static void
	parse_local_options_v2(const pcmk__client_t *cib_client,
	const cib_operation_t *operation, int call_options,
	const char host, const char op, gboolean *local_notify,
	gboolean needs_reply, gboolean process,
	gboolean *needs_forward)
	{
	// Process locally and notify local client
	*process = TRUE;
	*needs_reply = FALSE;
	*local_notify = TRUE;
	*needs_forward = FALSE;

	if (pcmk_is_set(operation->flags, cib_op_attr_local)) {
	/* Always process locally if cib_op_attr_local is set.
	*
	* @COMPAT: Currently host is ignored. At a compatibility break, throw
	* an error (from cib_process_request() or earlier) if host is not NULL or
	* OUR_NODENAME.
	*/
	crm_trace("Processing always-local %s op from client %s",
	op, pcmk__client_name(cib_client));

	if (!pcmk__str_eq(host, OUR_NODENAME,
	pcmk__str_casei\|pcmk__str_null_matches)) {

	crm_warn("Operation '%s' is always local but its target host is "
	"set to '%s'",
	op, host);
	}
	return;
	}

	if (pcmk_is_set(operation->flags, cib_op_attr_modifies)
	\|\| !pcmk__str_eq(host, OUR_NODENAME,
	pcmk__str_casei\|pcmk__str_null_matches)) {

	// Forward modifying and non-local requests via cluster
	*process = FALSE;
	*needs_reply = FALSE;
	*local_notify = FALSE;
	*needs_forward = TRUE;

	crm_trace("%s op from %s needs to be forwarded to %s",
	op, pcmk__client_name(cib_client),
	pcmk__s(host, "all nodes"));
	return;
	}

	if (stand_alone) {
	crm_trace("Processing %s op from client %s (stand-alone)",
	op, pcmk__client_name(cib_client));

	} else {
	crm_trace("Processing %saddressed %s op from client %s",
	((host != NULL)? "locally " : "un"),
	op, pcmk__client_name(cib_client));
	}
	}

	static void
	parse_local_options(const pcmk__client_t *cib_client,
	const cib_operation_t *operation, int call_options,
	const char host, const char op, gboolean *local_notify,
	gboolean needs_reply, gboolean process,
	gboolean *needs_forward)
	{
	if (pcmk_is_set(call_options, cib_transaction)) {
	/* Process locally and don't notify local client.
	*
	* Reaching cib_process_request() with cib_transaction set means we're
	* committing a transaction (always local and atomic).
	*/
	*process = TRUE;
	*needs_reply = FALSE;
	*local_notify = FALSE;
	*needs_forward = FALSE;
	return;
	}

	if(cib_legacy_mode()) {
	parse_local_options_v1(cib_client, operation, call_options, host,
	op, local_notify, needs_reply, process,
	needs_forward);
	} else {
	parse_local_options_v2(cib_client, operation, call_options, host,
	op, local_notify, needs_reply, process,
	needs_forward);
	}
	}

	static gboolean
	parse_peer_options_v1(const cib_operation_t operation, xmlNode request,
	gboolean local_notify, gboolean needs_reply,
	gboolean *process)
	{
	const char *op = NULL;
	const char *host = NULL;
	const char *delegated = NULL;
	const char *originator = crm_element_value(request, F_ORIG);
	const char *reply_to = crm_element_value(request, F_CIB_ISREPLY);

	gboolean is_reply = pcmk__str_eq(reply_to, OUR_NODENAME, pcmk__str_casei);

	if (pcmk__xe_attr_is_true(request, F_CIB_GLOBAL_UPDATE)) {
	*needs_reply = FALSE;
	if (is_reply) {
	*local_notify = TRUE;
	crm_trace("Processing global/peer update from %s"
	" that originated from us", originator);
	} else {
	crm_trace("Processing global/peer update from %s", originator);
	}
	return TRUE;
	}

	op = crm_element_value(request, F_CIB_OPERATION);
	crm_trace("Processing %s request sent by %s", op, originator);
	if (pcmk__str_eq(op, PCMK__CIB_REQUEST_SHUTDOWN, pcmk__str_none)) {
	/* Always process these */
	*local_notify = FALSE;
	if (reply_to == NULL \|\| is_reply) {
	*process = TRUE;
	}
	if (is_reply) {
	*needs_reply = FALSE;
	}
	return *process;
	}

	if (is_reply && pcmk__str_eq(op, CRM_OP_PING, pcmk__str_casei)) {
	process_ping_reply(request);
	return FALSE;
	}

	if (is_reply) {
	crm_trace("Forward reply sent from %s to local clients", originator);
	*process = FALSE;
	*needs_reply = FALSE;
	*local_notify = TRUE;
	return TRUE;
	}

	host = crm_element_value(request, F_CIB_HOST);
	if (pcmk__str_eq(host, OUR_NODENAME, pcmk__str_casei)) {
	crm_trace("Processing %s request sent to us from %s", op, originator);
	return TRUE;

	} else if(is_reply == FALSE && pcmk__str_eq(op, CRM_OP_PING, pcmk__str_casei)) {
	crm_trace("Processing %s request sent to %s by %s", op, host?host:"everyone", originator);
	*needs_reply = TRUE;
	return TRUE;

	} else if ((host == NULL) && based_is_primary) {
	crm_trace("Processing %s request sent to primary instance from %s",
	op, originator);
	return TRUE;
	}

	delegated = crm_element_value(request, F_CIB_DELEGATED);
	if (delegated != NULL) {
	crm_trace("Ignoring message for primary instance");

	} else if (host != NULL) {
	/* this is for a specific instance and we're not it */
	crm_trace("Ignoring msg for instance on %s", host);

	} else if ((reply_to == NULL) && !based_is_primary) {
	// This is for the primary instance, and we're not it
	crm_trace("Ignoring reply for primary instance");

	} else if (pcmk__str_eq(op, PCMK__CIB_REQUEST_SHUTDOWN, pcmk__str_none)) {
	if (reply_to != NULL) {
	crm_debug("Processing %s from %s", op, originator);
	*needs_reply = FALSE;

	} else {
	crm_debug("Processing %s reply from %s", op, originator);
	}
	return TRUE;

	} else {
	crm_err("Nothing for us to do?");
	crm_log_xml_err(request, "Peer[inbound]");
	}

	return FALSE;
	}

	static gboolean
	parse_peer_options_v2(const cib_operation_t operation, xmlNode request,
	gboolean local_notify, gboolean needs_reply,
	gboolean *process)
	{
	const char *host = NULL;
	const char *delegated = crm_element_value(request, F_CIB_DELEGATED);
	const char *op = crm_element_value(request, F_CIB_OPERATION);
	const char *originator = crm_element_value(request, F_ORIG);
	const char *reply_to = crm_element_value(request, F_CIB_ISREPLY);

	gboolean is_reply = pcmk__str_eq(reply_to, OUR_NODENAME, pcmk__str_casei);

	if (pcmk__str_eq(op, PCMK__CIB_REQUEST_REPLACE, pcmk__str_none)) {
	/* sync_our_cib() sets F_CIB_ISREPLY */
	if (reply_to) {
	delegated = reply_to;
	}
	goto skip_is_reply;

	} else if (pcmk__str_eq(op, PCMK__CIB_REQUEST_SYNC_TO_ALL,
	pcmk__str_none)) {
	// Nothing to do

	} else if (is_reply && pcmk__str_eq(op, CRM_OP_PING, pcmk__str_casei)) {
	process_ping_reply(request);
	return FALSE;

	} else if (pcmk__str_eq(op, PCMK__CIB_REQUEST_UPGRADE, pcmk__str_none)) {
	/* Only the DC (node with the oldest software) should process
	* this operation if F_CIB_SCHEMA_MAX is unset
	*
	* If the DC is happy it will then send out another
	* PCMK__CIB_REQUEST_UPGRADE which will tell all nodes to do the actual
	* upgrade.
	*
	* Except this time F_CIB_SCHEMA_MAX will be set which puts a
	* limit on how far newer nodes will go
	*/
	const char *max = crm_element_value(request, F_CIB_SCHEMA_MAX);
	const char *upgrade_rc = crm_element_value(request, F_CIB_UPGRADE_RC);

	crm_trace("Parsing %s operation%s for %s with max=%s and upgrade_rc=%s",
	op, (is_reply? " reply" : ""),
	(based_is_primary? "primary" : "secondary"),
	(max? max : "none"), (upgrade_rc? upgrade_rc : "none"));

	if (upgrade_rc != NULL) {
	// Our upgrade request was rejected by DC, notify clients of result
	crm_xml_add(request, F_CIB_RC, upgrade_rc);

	} else if ((max == NULL) && based_is_primary) {
	/* We are the DC, check if this upgrade is allowed */
	goto skip_is_reply;

	} else if(max) {
	/* Ok, go ahead and upgrade to 'max' */
	goto skip_is_reply;

	} else {
	// Ignore broadcast client requests when we're not DC
	return FALSE;
	}

	} else if (pcmk__xe_attr_is_true(request, F_CIB_GLOBAL_UPDATE)) {
	crm_info("Detected legacy %s global update from %s", op, originator);
	send_sync_request(NULL);
	legacy_mode = TRUE;
	return FALSE;

	} else if (is_reply
	&& pcmk_is_set(operation->flags, cib_op_attr_modifies)) {
	crm_trace("Ignoring legacy %s reply sent from %s to local clients", op, originator);
	return FALSE;

	} else if (pcmk__str_eq(op, PCMK__CIB_REQUEST_SHUTDOWN, pcmk__str_none)) {
	// @COMPAT: Legacy handling
	crm_debug("Legacy handling of %s message from %s", op, originator);
	*local_notify = FALSE;
	if (reply_to == NULL) {
	*process = TRUE;
	}
	return *process;
	}

	if(is_reply) {
	crm_trace("Handling %s reply sent from %s to local clients", op, originator);
	*process = FALSE;
	*needs_reply = FALSE;
	*local_notify = TRUE;
	return TRUE;
	}

	skip_is_reply:
	*process = TRUE;
	*needs_reply = FALSE;

	*local_notify = pcmk__str_eq(delegated, OUR_NODENAME, pcmk__str_casei);

	host = crm_element_value(request, F_CIB_HOST);
	if (pcmk__str_eq(host, OUR_NODENAME, pcmk__str_casei)) {
	crm_trace("Processing %s request sent to us from %s", op, originator);
	*needs_reply = TRUE;
	return TRUE;

	} else if (host != NULL) {
	/* this is for a specific instance and we're not it */
	crm_trace("Ignoring %s operation for instance on %s", op, host);
	return FALSE;

	} else if(is_reply == FALSE && pcmk__str_eq(op, CRM_OP_PING, pcmk__str_casei)) {
	*needs_reply = TRUE;
	}

	crm_trace("Processing %s request sent to everyone by %s/%s on %s %s", op,
	crm_element_value(request, F_CIB_CLIENTNAME),
	crm_element_value(request, F_CIB_CALLID),
	originator, (*local_notify)?"(notify)":"");
	return TRUE;
	}

	static gboolean
	parse_peer_options(const cib_operation_t operation, xmlNode request,
	gboolean local_notify, gboolean needs_reply,
	gboolean *process)
	{
	/* TODO: What happens when an update comes in after node A
	* requests the CIB from node B, but before it gets the reply (and
	* sends out the replace operation)
	*/
	if(cib_legacy_mode()) {
	return parse_peer_options_v1(operation, request, local_notify,
	needs_reply, process);
	} else {
	return parse_peer_options_v2(operation, request, local_notify,
	needs_reply, process);
	}
	}

	/*!
	* \internal
	* \brief Forward a CIB request to the appropriate target host(s)
	*
	* \param[in] request CIB request to forward
	*/
	static void
	forward_request(xmlNode *request)
	{
	const char *op = crm_element_value(request, F_CIB_OPERATION);
	const char *section = crm_element_value(request, F_CIB_SECTION);
	const char *host = crm_element_value(request, F_CIB_HOST);
	const char *originator = crm_element_value(request, F_ORIG);
	const char *client_name = crm_element_value(request, F_CIB_CLIENTNAME);
	const char *call_id = crm_element_value(request, F_CIB_CALLID);

	int log_level = LOG_INFO;

	if (pcmk__str_eq(op, PCMK__CIB_REQUEST_NOOP, pcmk__str_none)) {
	log_level = LOG_DEBUG;
	}

	do_crm_log(log_level,
	"Forwarding %s operation for section %s to %s (origin=%s/%s/%s)",
	pcmk__s(op, "invalid"),
	pcmk__s(section, "all"),
	pcmk__s(host, (cib_legacy_mode()? "primary" : "all")),
	pcmk__s(originator, "local"),
	pcmk__s(client_name, "unspecified"),
	pcmk__s(call_id, "unspecified"));

	crm_xml_add(request, F_CIB_DELEGATED, OUR_NODENAME);

	send_cluster_message(((host != NULL)? crm_get_peer(0, host) : NULL),
	crm_msg_cib, request, FALSE);

	// Return the request to its original state
	xml_remove_prop(request, F_CIB_DELEGATED);
	}

	static gboolean
	send_peer_reply(xmlNode * msg, xmlNode * result_diff, const char *originator, gboolean broadcast)
	{
	CRM_ASSERT(msg != NULL);

	if (broadcast) {
	/* @COMPAT: Legacy code
	*
	* This successful call modified the CIB, and the change needs to be
	* broadcast (sent via cluster to all nodes).
	*/
	int diff_add_updates = 0;
	int diff_add_epoch = 0;
	int diff_add_admin_epoch = 0;

	int diff_del_updates = 0;
	int diff_del_epoch = 0;
	int diff_del_admin_epoch = 0;

	const char *digest = NULL;
	int format = 1;

	CRM_LOG_ASSERT(result_diff != NULL);
	digest = crm_element_value(result_diff, XML_ATTR_DIGEST);
	crm_element_value_int(result_diff, "format", &format);

	cib_diff_version_details(result_diff,
	&diff_add_admin_epoch, &diff_add_epoch, &diff_add_updates,
	&diff_del_admin_epoch, &diff_del_epoch, &diff_del_updates);

	crm_trace("Sending update diff %d.%d.%d -> %d.%d.%d %s",
	diff_del_admin_epoch, diff_del_epoch, diff_del_updates,
	diff_add_admin_epoch, diff_add_epoch, diff_add_updates, digest);

	crm_xml_add(msg, F_CIB_ISREPLY, originator);
	pcmk__xe_set_bool_attr(msg, F_CIB_GLOBAL_UPDATE, true);
	crm_xml_add(msg, F_CIB_OPERATION, PCMK__CIB_REQUEST_APPLY_PATCH);
	crm_xml_add(msg, F_CIB_USER, CRM_DAEMON_USER);

	if (format == 1) {
	CRM_ASSERT(digest != NULL);
	}

	add_message_xml(msg, F_CIB_UPDATE_DIFF, result_diff);
	crm_log_xml_explicit(msg, "copy");
	return send_cluster_message(NULL, crm_msg_cib, msg, TRUE);

	} else if (originator != NULL) {
	/* send reply via HA to originating node */
	crm_trace("Sending request result to %s only", originator);
	crm_xml_add(msg, F_CIB_ISREPLY, originator);
	return send_cluster_message(crm_get_peer(0, originator), crm_msg_cib, msg, FALSE);
	}

	return FALSE;
	}

	/*!
	* \internal
	* \brief Handle an IPC or CPG message containing a request
	*
	* \param[in,out] request Request XML
	* \param[in] privileged Whether privileged commands may be run
	* (see cib_server_ops[] definition)
	* \param[in] cib_client IPC client that sent request (or NULL if CPG)
	*
	* \return Legacy Pacemaker return code
	*/
	int
	cib_process_request(xmlNode *request, gboolean privileged,
	const pcmk__client_t *cib_client)
	{
	// @TODO: Break into multiple smaller functions
	int call_options = 0;

	gboolean process = TRUE; // Whether to process request locally now
	gboolean is_update = TRUE; // Whether request would modify CIB
	gboolean needs_reply = TRUE; // Whether to build a reply
	gboolean local_notify = FALSE; // Whether to notify (local) requester
	gboolean needs_forward = FALSE; // Whether to forward request somewhere else

	xmlNode *op_reply = NULL;
	xmlNode *result_diff = NULL;

	int rc = pcmk_ok;
	const char *op = crm_element_value(request, F_CIB_OPERATION);
	const char *originator = crm_element_value(request, F_ORIG);
	const char *host = crm_element_value(request, F_CIB_HOST);
	const char *target = NULL;
	const char *call_id = crm_element_value(request, F_CIB_CALLID);
	const char *client_id = crm_element_value(request, F_CIB_CLIENTID);
	const char *client_name = crm_element_value(request, F_CIB_CLIENTNAME);
	const char *reply_to = crm_element_value(request, F_CIB_ISREPLY);

	const cib_operation_t *operation = NULL;
	+ cib__op_fn_t op_function = NULL;

	crm_element_value_int(request, F_CIB_CALLOPTS, &call_options);

	if ((host != NULL) && (*host == '\0')) {
	host = NULL;
	}

	// @TODO: Improve trace messages. Target is accurate only for legacy mode.
	if (host) {
	target = host;

	} else if (call_options & cib_scope_local) {
	target = "local host";

	} else {
	target = "primary";
	}

	if (cib_client == NULL) {
	crm_trace("Processing peer %s operation from %s/%s on %s intended for %s (reply=%s)",
	op, client_name, call_id, originator, target, reply_to);
	} else {
	crm_xml_add(request, F_ORIG, OUR_NODENAME);
	crm_trace("Processing local %s operation from %s/%s intended for %s", op, client_name, call_id, target);
	}

	- rc = cib_get_operation(op, &operation);
	+ rc = cib__get_operation(op, &operation);
	+ rc = pcmk_rc2legacy(rc);
	if (rc != pcmk_ok) {
	/* TODO: construct error reply? */
	crm_err("Pre-processing of command failed: %s", pcmk_strerror(rc));
	return rc;
	}

	+ op_function = based_get_op_function(operation);
	+ if (op_function == NULL) {
	+ crm_err("Operation %s not supported by CIB manager", op);
	+ return -EOPNOTSUPP;
	+ }
	+
	if (pcmk_is_set(call_options, cib_transaction)) {
	/* We're committing a transaction now, processing each request. If
	* cib_op_attr_transaction is not set, based_extend_transaction() should
	* have blocked the request from being added to the transaction.
	*/
	CRM_CHECK(pcmk_is_set(operation->flags, cib_op_attr_transaction),
	return -EOPNOTSUPP);
	}

	if (cib_client != NULL) {
	parse_local_options(cib_client, operation, call_options, host, op,
	&local_notify, &needs_reply, &process,
	&needs_forward);

	} else if (!parse_peer_options(operation, request, &local_notify,
	&needs_reply, &process)) {
	return rc;
	}

	is_update = pcmk_is_set(operation->flags, cib_op_attr_modifies);

	if (pcmk_is_set(call_options, cib_discard_reply)) {
	/* If the request will modify the CIB, and we are in legacy mode, we
	* need to build a reply so we can broadcast a diff, even if the
	* requester doesn't want one.
	*/
	needs_reply = is_update && cib_legacy_mode();
	local_notify = FALSE;
	crm_trace("Client is not interested in the reply");
	}

	if (needs_forward) {
	forward_request(request);
	return rc;
	}

	if (cib_status != pcmk_ok) {
	rc = cib_status;
	crm_err("Operation ignored, cluster configuration is invalid."
	" Please repair and restart: %s", pcmk_strerror(cib_status));

	op_reply = create_cib_reply(op, call_id, client_id, call_options, rc,
	the_cib);

	} else if (process) {
	time_t finished = 0;
	time_t now = time(NULL);
	int level = LOG_INFO;
	const char *section = crm_element_value(request, F_CIB_SECTION);

	- rc = cib_process_command(request, operation, &op_reply, &result_diff,
	- privileged);
	+ rc = cib_process_command(request, operation, op_function, &op_reply,
	+ &result_diff, privileged);

	if (!is_update) {
	level = LOG_TRACE;

	} else if (pcmk__xe_attr_is_true(request, F_CIB_GLOBAL_UPDATE)) {
	switch (rc) {
	case pcmk_ok:
	level = LOG_INFO;
	break;
	case -pcmk_err_old_data:
	case -pcmk_err_diff_resync:
	case -pcmk_err_diff_failed:
	level = LOG_TRACE;
	break;
	default:
	level = LOG_ERR;
	}

	} else if (rc != pcmk_ok) {
	level = LOG_WARNING;
	}

	do_crm_log(level,
	"Completed %s operation for section %s: %s (rc=%d, origin=%s/%s/%s, version=%s.%s.%s)",
	op, section ? section : "'all'", pcmk_strerror(rc), rc,
	originator ? originator : "local", client_name, call_id,
	the_cib ? crm_element_value(the_cib, XML_ATTR_GENERATION_ADMIN) : "0",
	the_cib ? crm_element_value(the_cib, XML_ATTR_GENERATION) : "0",
	the_cib ? crm_element_value(the_cib, XML_ATTR_NUMUPDATES) : "0");

	finished = time(NULL);
	if ((finished - now) > 3) {
	crm_trace("%s operation took %lds to complete", op, (long)(finished - now));
	crm_write_blackbox(0, NULL);
	}

	if (op_reply == NULL && (needs_reply \|\| local_notify)) {
	crm_err("Unexpected NULL reply to message");
	crm_log_xml_err(request, "null reply");
	needs_reply = FALSE;
	local_notify = FALSE;
	}
	}

	if (is_update && !cib_legacy_mode()) {
	crm_trace("Completed pre-sync update from %s/%s/%s%s",
	originator ? originator : "local", client_name, call_id,
	local_notify?" with local notification":"");

	} else if (!needs_reply \|\| stand_alone) {
	// This was a non-originating secondary update
	crm_trace("Completed update as secondary");

	} else if (cib_legacy_mode() &&
	rc == pcmk_ok && result_diff != NULL && !(call_options & cib_inhibit_bcast)) {
	gboolean broadcast = FALSE;

	cib_local_bcast_num++;
	crm_xml_add_int(request, F_CIB_LOCAL_NOTIFY_ID, cib_local_bcast_num);
	broadcast = send_peer_reply(request, result_diff, originator, TRUE);

	if (broadcast && client_id && local_notify && op_reply) {

	/* If we have been asked to sync the reply,
	* and a bcast msg has gone out, we queue the local notify
	* until we know the bcast message has been received */
	local_notify = FALSE;
	crm_trace("Queuing local %ssync notification for %s",
	(call_options & cib_sync_call) ? "" : "a-", client_id);

	queue_local_notify(op_reply, client_id,
	pcmk_is_set(call_options, cib_sync_call),
	(cib_client == NULL));
	op_reply = NULL; /* the reply is queued, so don't free here */
	}

	} else if ((cib_client == NULL)
	&& !pcmk_is_set(call_options, cib_discard_reply)) {

	if (is_update == FALSE \|\| result_diff == NULL) {
	crm_trace("Request not broadcast: R/O call");

	} else if (call_options & cib_inhibit_bcast) {
	crm_trace("Request not broadcast: inhibited");

	} else if (rc != pcmk_ok) {
	crm_trace("Request not broadcast: call failed: %s", pcmk_strerror(rc));

	} else {
	crm_trace("Directing reply to %s", originator);
	}

	send_peer_reply(op_reply, result_diff, originator, FALSE);
	}

	if (local_notify && client_id) {
	crm_trace("Performing local %ssync notification for %s",
	(pcmk_is_set(call_options, cib_sync_call)? "" : "a"),
	client_id);
	if (process == FALSE) {
	do_local_notify(request, client_id,
	pcmk_is_set(call_options, cib_sync_call),
	(cib_client == NULL));
	} else {
	do_local_notify(op_reply, client_id,
	pcmk_is_set(call_options, cib_sync_call),
	(cib_client == NULL));
	}
	}

	free_xml(op_reply);
	free_xml(result_diff);

	return rc;
	}

	/*!
	* \internal
	* \brief Get a CIB operation's input from the request XML
	*
	* \param[in] request CIB request XML
	* \param[in] type CIB operation type
	* \param[out] section Where to store CIB section name
	*
	* \return Input XML for CIB operation
	*
	* \note If not \c NULL, the return value is a non-const pointer to part of
	* \p request. The caller should not free it directly.
	*/
	static xmlNode *
	prepare_input(const xmlNode *request, enum cib__op_type type,
	const char **section)
	{
	xmlNode *root = NULL;

	if (type == cib__op_apply_patch) {
	if (pcmk__xe_attr_is_true(request, F_CIB_GLOBAL_UPDATE)) {
	root = get_message_xml(request, F_CIB_UPDATE_DIFF);
	} else {
	root = get_message_xml(request, F_CIB_CALLDATA);
	}
	*section = NULL;

	} else {
	root = get_message_xml(request, F_CIB_CALLDATA);
	*section = crm_element_value(request, F_CIB_SECTION);
	}

	if (root == NULL) {
	return NULL;
	}

	if (pcmk__str_any_of(crm_element_name(root), F_CRM_DATA, F_CIB_CALLDATA,
	NULL)) {
	root = first_named_child(root, XML_TAG_CIB);
	}

	// Grab the specified section
	if ((root != NULL) && (*section != NULL)
	&& pcmk__str_eq(crm_element_name(root), XML_TAG_CIB, pcmk__str_none)) {

	root = pcmk_find_cib_element(root, *section);
	}

	return root;
	}

	static char *
	calculate_section_digest(const char xpath, xmlNode xml_obj)
	{
	xmlNode *xml_section = NULL;

	if (xml_obj == NULL) {
	return NULL;
	}

	xml_section = get_xpath_object(xpath, xml_obj, LOG_TRACE);
	if (xml_section == NULL) {
	return NULL;
	}
	return calculate_xml_versioned_digest(xml_section, FALSE, TRUE, CRM_FEATURE_SET);

	}

	#define XPATH_CONFIG "/" XML_TAG_CIB "/" XML_CIB_TAG_CONFIGURATION
	#define XPATH_NODES XPATH_CONFIG "/" XML_CIB_TAG_NODES
	#define XPATH_ALERTS XPATH_CONFIG "/" XML_CIB_TAG_ALERTS
	#define XPATH_STATUS "/" XML_TAG_CIB "/" XML_CIB_TAG_STATUS

	/*!
	* \internal
	* \brief Calculate digests of CIB sections
	*
	* \param[in] cib CIB to get digests from
	* \param[out] digests Where to store digests
	*
	* \note The caller is responsible for freeing the output argument's members
	* using \p free_digests().
	*/
	static void
	get_digests(xmlNode cib, struct digest_data digests)
	{
	digests->nodes = calculate_section_digest(XPATH_NODES, cib);
	digests->alerts = calculate_section_digest(XPATH_ALERTS, cib);
	digests->status = calculate_section_digest(XPATH_STATUS, cib);
	}

	/*!
	* \internal
	* \brief Determine which CIB sections were changed by an operation
	*
	* \param[in] before Digests before the operation
	* \param[in] after Digests after the operation
	*
	* \return Group of <tt>enum cib_change_section_info</tt> flags indicating which
	* sections have changed
	*/
	static uint32_t
	get_change_sections(const struct digest_data *before,
	const struct digest_data *after)
	{
	uint32_t change_sections = cib_change_section_none;

	if (!pcmk__str_eq(before->nodes, after->nodes, pcmk__str_none)) {
	pcmk__set_change_section(change_sections, cib_change_section_nodes);
	}
	if (!pcmk__str_eq(before->alerts, after->alerts, pcmk__str_none)) {
	pcmk__set_change_section(change_sections, cib_change_section_alerts);
	}
	if (!pcmk__str_eq(before->status, after->status, pcmk__str_none)) {
	pcmk__set_change_section(change_sections, cib_change_section_status);
	}

	return change_sections;
	}

	/*!
	* \internal
	* \brief Check whether a CIB replace notification should be sent
	*
	* A CIB replace notification should be sent after an operation if the operation
	* is of an appropriate type, notifications are not disabled, and any of certain
	* CIB sections has changed.
	*
	* \param[in] before_digests Digests before operation
	* \param[in] after_cib Result CIB after operation
	* \param[in] operation CIB operation
	* \param[in] call_options Group of <tt>enum cib_call_options</tt> flags
	* \param[out] change_sections Group of <tt>enum cib_change_section_info</tt>
	* flags indicating which sections have changed
	*
	* \return \p true if a replace notification should be sent, or \p false
	* otherwise
	*/
	static bool
	should_replace_notify(const struct digest_data *before_digests,
	xmlNode after_cib, const cib_operation_t operation,
	int call_options, uint32_t *change_sections)
	{
	struct digest_data after_digests = { 0, };

	*change_sections = cib_change_section_none;

	if (!pcmk_is_set(operation->flags, cib_op_attr_replaces)
	\|\| pcmk_is_set(call_options, cib_inhibit_notify)) {
	return false;
	}

	get_digests(after_cib, &after_digests);
	crm_trace("after-digest %s:%s:%s",
	pcmk__s(after_digests.nodes, "(null)"),
	pcmk__s(after_digests.alerts, "(null)"),
	pcmk__s(after_digests.status, "(null)"));

	*change_sections = get_change_sections(before_digests, &after_digests);
	free_digests(&after_digests);

	return (*change_sections != cib_change_section_none);
	}

	// v1 and v2 patch formats
	#define XPATH_CONFIG_CHANGE \
	"//" XML_CIB_TAG_CRMCONFIG " \| " \
	"//" XML_DIFF_CHANGE \
	"[contains(@" XML_DIFF_PATH ",'/" XML_CIB_TAG_CRMCONFIG "/')]"

	static bool
	contains_config_change(xmlNode *diff)
	{
	bool changed = false;

	if (diff) {
	xmlXPathObject *xpathObj = xpath_search(diff, XPATH_CONFIG_CHANGE);

	if (numXpathResults(xpathObj) > 0) {
	changed = true;
	}
	freeXpathObject(xpathObj);
	}
	return changed;
	}

	static int
	cib_process_command(xmlNode request, const cib_operation_t operation,
	- xmlNode reply, xmlNode cib_diff, bool privileged)
	+ cib__op_fn_t op_function, xmlNode **reply,
	+ xmlNode **cib_diff, bool privileged)
	{
	xmlNode *input = NULL;
	xmlNode *output = NULL;
	xmlNode *result_cib = NULL;

	int call_options = 0;

	const char *op = NULL;
	const char *section = NULL;
	const char *call_id = crm_element_value(request, F_CIB_CALLID);
	const char *client_id = crm_element_value(request, F_CIB_CLIENTID);
	const char *client_name = crm_element_value(request, F_CIB_CLIENTNAME);
	const char *origin = crm_element_value(request, F_ORIG);

	int rc = pcmk_ok;

	bool config_changed = false;
	bool manage_counters = true;

	static mainloop_timer_t *digest_timer = NULL;

	struct digest_data before_digests = { 0, };

	CRM_ASSERT(cib_status == pcmk_ok);

	if(digest_timer == NULL) {
	digest_timer = mainloop_timer_add("digester", 5000, FALSE, cib_digester_cb, NULL);
	}

	*reply = NULL;
	*cib_diff = NULL;

	/* Start processing the request... */
	op = crm_element_value(request, F_CIB_OPERATION);
	crm_element_value_int(request, F_CIB_CALLOPTS, &call_options);

	if (!privileged && pcmk_is_set(operation->flags, cib_op_attr_privileged)) {
	rc = -EACCES;
	crm_trace("Failed due to lack of privileges: %s", pcmk_strerror(rc));
	goto done;
	}

	input = prepare_input(request, operation->type, &section);

	if (!pcmk_is_set(operation->flags, cib_op_attr_modifies)) {
	- rc = cib_perform_op(op, call_options, operation->fn, true, section,
	+ rc = cib_perform_op(op, call_options, op_function, true, section,
	request, input, false, &config_changed, &the_cib,
	&result_cib, NULL, &output);

	CRM_CHECK(result_cib == NULL, free_xml(result_cib));
	goto done;
	}

	/* @COMPAT: Handle a valid write action (legacy)
	*
	* @TODO: Re-evaluate whether this is all truly legacy. The cib_force_diff
	* portion is. However, F_CIB_GLOBAL_UPDATE may be set by a sync operation
	* even in non-legacy mode, and manage_counters tells xml_create_patchset()
	* whether to update version/epoch info.
	*/
	if (pcmk__xe_attr_is_true(request, F_CIB_GLOBAL_UPDATE)) {
	manage_counters = false;
	cib__set_call_options(call_options, "call", cib_force_diff);
	crm_trace("Global update detected");

	CRM_LOG_ASSERT(pcmk__str_any_of(op,
	PCMK__CIB_REQUEST_APPLY_PATCH,
	PCMK__CIB_REQUEST_REPLACE,
	NULL));
	}

	ping_modified_since = TRUE;
	if (pcmk_is_set(call_options, cib_inhibit_bcast)) {
	crm_trace("Skipping update: inhibit broadcast");
	manage_counters = false;
	}

	// Calculate the digests of relevant sections before the operation
	if (pcmk_is_set(operation->flags, cib_op_attr_replaces)
	&& !pcmk_any_flags_set(call_options,
	cib_dryrun\|cib_inhibit_notify\|cib_transaction)) {

	get_digests(the_cib, &before_digests);
	crm_trace("before-digest %s:%s:%s",
	pcmk__s(before_digests.nodes, "(null)"),
	pcmk__s(before_digests.alerts, "(null)"),
	pcmk__s(before_digests.status, "(null)"));
	}

	// result_cib must not be modified after cib_perform_op() returns
	- rc = cib_perform_op(op, call_options, operation->fn, false, section,
	+ rc = cib_perform_op(op, call_options, op_function, false, section,
	request, input, manage_counters, &config_changed,
	&the_cib, &result_cib, cib_diff, &output);

	// @COMPAT: Legacy code
	if (!manage_counters) {
	int format = 1;

	// If the diff is NULL at this point, it's because nothing changed
	if (*cib_diff != NULL) {
	crm_element_value_int(*cib_diff, "format", &format);
	}

	if (format == 1) {
	config_changed = cib__config_changed_v1(NULL, NULL, cib_diff);
	}
	}

	/* Always write to disk for successful ops with the flag set. This also
	* negates the need to detect ordering changes.
	*/
	if ((rc == pcmk_ok)
	&& pcmk_is_set(operation->flags, cib_op_attr_writes_through)) {

	config_changed = true;
	}

	if ((rc == pcmk_ok)
	&& !pcmk_any_flags_set(call_options, cib_dryrun\|cib_transaction)) {

	uint32_t change_sections = cib_change_section_none;

	if (result_cib != the_cib) {
	crm_trace("Activating %s->%s%s",
	crm_element_value(the_cib, XML_ATTR_NUMUPDATES),
	crm_element_value(result_cib, XML_ATTR_NUMUPDATES),
	(config_changed? " changed" : ""));

	rc = activateCibXml(result_cib, config_changed, op);
	if (rc != pcmk_ok) {
	crm_err("Failed to activate new CIB: %s", pcmk_strerror(rc));
	}
	}

	if ((rc == pcmk_ok) && contains_config_change(*cib_diff)) {
	cib_read_config(config_hash, result_cib);
	}

	if (should_replace_notify(&before_digests, result_cib, operation,
	call_options, &change_sections)) {

	// @TODO: Should update argument be result_cib instead of the_cib?
	cib_replace_notify(op, rc, call_id, client_id, client_name, origin,
	the_cib, *cib_diff, change_sections);
	}

	/* Commit-transaction is special. We process it and activate the result
	* only locally, but we still need to sync the updated CIB to all nodes.
	* However, if we run the sync within cib_process_commit_transaction(),
	* it will be rejected locally due to an older epoch in the replacement
	* CIB. cib_perform_op() updates the epoch via xml_create_patchset()
	* after cib_process_commit_transaction() has already returned.
	*/
	if (pcmk__str_eq(op, PCMK__CIB_REQUEST_COMMIT_TRANSACT,
	pcmk__str_none)) {
	sync_our_cib(request, TRUE);
	}

	mainloop_timer_stop(digest_timer);
	mainloop_timer_start(digest_timer);

	} else if (rc == -pcmk_err_schema_validation) {
	CRM_ASSERT(result_cib != the_cib);

	if (output != NULL) {
	crm_log_xml_info(output, "cib:output");
	free_xml(output);
	}

	output = result_cib;

	} else {
	crm_trace("Not activating %d %d %s", rc,
	pcmk_is_set(call_options, cib_dryrun),
	crm_element_value(result_cib, XML_ATTR_NUMUPDATES));

	if (result_cib != the_cib) {
	free_xml(result_cib);
	}
	}

	if (!pcmk_any_flags_set(call_options,
	cib_dryrun\|cib_inhibit_notify\|cib_transaction)) {
	crm_trace("Sending notifications %d",
	pcmk_is_set(call_options, cib_dryrun));
	cib_diff_notify(op, rc, call_id, client_id, client_name, origin, input,
	*cib_diff);
	}

	pcmk__output_set_log_level(logger_out, LOG_TRACE);
	logger_out->message(logger_out, "xml-patchset", *cib_diff);

	done:
	if (!pcmk_is_set(call_options, cib_discard_reply) \|\| cib_legacy_mode()) {
	*reply = create_cib_reply(op, call_id, client_id, call_options, rc,
	output);
	}

	if (output != the_cib) {
	free_xml(output);
	}
	free_digests(&before_digests);
	crm_trace("done");
	return rc;
	}

	void
	cib_peer_callback(xmlNode * msg, void *private_data)
	{
	const char *reason = NULL;
	const char *originator = crm_element_value(msg, F_ORIG);

	if (cib_legacy_mode()
	&& pcmk__str_eq(originator, OUR_NODENAME,
	pcmk__str_casei\|pcmk__str_null_matches)) {
	/* message is from ourselves */
	int bcast_id = 0;

	if (!(crm_element_value_int(msg, F_CIB_LOCAL_NOTIFY_ID, &bcast_id))) {
	check_local_notify(bcast_id);
	}
	return;

	} else if (crm_peer_cache == NULL) {
	reason = "membership not established";
	goto bail;
	}

	if (crm_element_value(msg, F_CIB_CLIENTNAME) == NULL) {
	crm_xml_add(msg, F_CIB_CLIENTNAME, originator);
	}

	/* crm_log_xml_trace(msg, "Peer[inbound]"); */
	cib_process_request(msg, TRUE, NULL);
	return;

	bail:
	if (reason) {
	const char *seq = crm_element_value(msg, F_SEQ);
	const char *op = crm_element_value(msg, F_CIB_OPERATION);

	crm_warn("Discarding %s message (%s) from %s: %s", op, seq, originator, reason);
	}
	}

	static gboolean
	cib_force_exit(gpointer data)
	{
	crm_notice("Forcing exit!");
	terminate_cib(__func__, CRM_EX_ERROR);
	return FALSE;
	}

	static void
	disconnect_remote_client(gpointer key, gpointer value, gpointer user_data)
	{
	pcmk__client_t *a_client = value;

	crm_err("Can't disconnect client %s: Not implemented",
	pcmk__client_name(a_client));
	}

	static void
	initiate_exit(void)
	{
	int active = 0;
	xmlNode *leaving = NULL;

	active = crm_active_peers();
	if (active < 2) {
	terminate_cib(__func__, 0);
	return;
	}

	crm_info("Sending disconnect notification to %d peers...", active);

	leaving = create_xml_node(NULL, "exit-notification");
	crm_xml_add(leaving, F_TYPE, "cib");
	crm_xml_add(leaving, F_CIB_OPERATION, PCMK__CIB_REQUEST_SHUTDOWN);

	send_cluster_message(NULL, crm_msg_cib, leaving, TRUE);
	free_xml(leaving);

	g_timeout_add(EXIT_ESCALATION_MS, cib_force_exit, NULL);
	}

	void
	cib_shutdown(int nsig)
	{
	struct qb_ipcs_stats srv_stats;

	if (cib_shutdown_flag == FALSE) {
	int disconnects = 0;
	qb_ipcs_connection_t *c = NULL;

	cib_shutdown_flag = TRUE;

	c = qb_ipcs_connection_first_get(ipcs_rw);
	while (c != NULL) {
	qb_ipcs_connection_t *last = c;

	c = qb_ipcs_connection_next_get(ipcs_rw, last);

	crm_debug("Disconnecting r/w client %p...", last);
	qb_ipcs_disconnect(last);
	qb_ipcs_connection_unref(last);
	disconnects++;
	}

	c = qb_ipcs_connection_first_get(ipcs_ro);
	while (c != NULL) {
	qb_ipcs_connection_t *last = c;

	c = qb_ipcs_connection_next_get(ipcs_ro, last);

	crm_debug("Disconnecting r/o client %p...", last);
	qb_ipcs_disconnect(last);
	qb_ipcs_connection_unref(last);
	disconnects++;
	}

	c = qb_ipcs_connection_first_get(ipcs_shm);
	while (c != NULL) {
	qb_ipcs_connection_t *last = c;

	c = qb_ipcs_connection_next_get(ipcs_shm, last);

	crm_debug("Disconnecting non-blocking r/w client %p...", last);
	qb_ipcs_disconnect(last);
	qb_ipcs_connection_unref(last);
	disconnects++;
	}

	disconnects += pcmk__ipc_client_count();

	crm_debug("Disconnecting %d remote clients", pcmk__ipc_client_count());
	pcmk__foreach_ipc_client(disconnect_remote_client, NULL);
	crm_info("Disconnected %d clients", disconnects);
	}

	qb_ipcs_stats_get(ipcs_rw, &srv_stats, QB_FALSE);

	if (pcmk__ipc_client_count() == 0) {
	crm_info("All clients disconnected (%d)", srv_stats.active_connections);
	initiate_exit();

	} else {
	crm_info("Waiting on %d clients to disconnect (%d)",
	pcmk__ipc_client_count(), srv_stats.active_connections);
	}
	}

	extern int remote_fd;
	extern int remote_tls_fd;

	/*!
	* \internal
	* \brief Close remote sockets, free the global CIB and quit
	*
	* \param[in] caller Name of calling function (for log message)
	* \param[in] fast If -1, skip disconnect; if positive, exit that
	*/
	void
	terminate_cib(const char *caller, int fast)
	{
	crm_info("%s: Exiting%s...", caller,
	(fast > 0)? " fast" : mainloop ? " from mainloop" : "");

	if (remote_fd > 0) {
	close(remote_fd);
	remote_fd = 0;
	}
	if (remote_tls_fd > 0) {
	close(remote_tls_fd);
	remote_tls_fd = 0;
	}

	uninitializeCib();

	if (logger_out != NULL) {
	logger_out->finish(logger_out, CRM_EX_OK, true, NULL);
	pcmk__output_free(logger_out);
	logger_out = NULL;
	}

	if (fast > 0) {
	/* Quit fast on error */
	pcmk__stop_based_ipc(ipcs_ro, ipcs_rw, ipcs_shm);
	crm_exit(fast);

	} else if ((mainloop != NULL) && g_main_loop_is_running(mainloop)) {
	/* Quit via returning from the main loop. If fast == -1, we skip the
	* disconnect here, and it will be done when the main loop returns
	* (this allows the peer status callback to avoid messing with the
	* peer caches).
	*/
	if (fast == 0) {
	crm_cluster_disconnect(crm_cluster);
	}
	g_main_loop_quit(mainloop);

	} else {
	/* Quit via clean exit. Even the peer status callback can disconnect
	* here, because we're not returning control to the caller. */
	crm_cluster_disconnect(crm_cluster);
	pcmk__stop_based_ipc(ipcs_ro, ipcs_rw, ipcs_shm);
	crm_exit(CRM_EX_OK);
	}
	}
	diff --git a/daemons/based/based_operation.c b/daemons/based/based_operation.c
	index 38fe086b06..78a79c77f3 100644
	--- a/daemons/based/based_operation.c
	+++ b/daemons/based/based_operation.c
	@@ -1,187 +1,82 @@
	/*
	* Copyright 2008-2023 the Pacemaker project contributors
	*
	* The version control history for this file may have further details.
	*
	* This source code is licensed under the GNU General Public License version 2
	* or later (GPLv2+) WITHOUT ANY WARRANTY.
	*/

	#include <crm_internal.h>

	#include <sys/param.h>
	#include <stdio.h>
	#include <sys/types.h>
	#include <unistd.h>

	#include <stdlib.h>
	#include <errno.h>
	#include <fcntl.h>

	#include <glib.h>
	#include <libxml/tree.h>

	#include <crm/crm.h>
	#include <crm/cib.h>
	#include <crm/msg_xml.h>
	#include <crm/common/ipc.h>
	#include <crm/cluster.h>

	#include <crm/common/xml.h>

	#include <pacemaker-based.h>

	-static const cib_operation_t cib_server_ops[] = {
	- {
	- PCMK__CIB_REQUEST_QUERY, cib__op_query,
	- cib_op_attr_none,
	- cib_process_query
	- },
	- {
	- PCMK__CIB_REQUEST_MODIFY, cib__op_modify,
	- cib_op_attr_modifies\|cib_op_attr_privileged\|cib_op_attr_transaction,
	- cib_process_modify
	- },
	- {
	- PCMK__CIB_REQUEST_APPLY_PATCH, cib__op_apply_patch,
	- cib_op_attr_modifies\|cib_op_attr_privileged\|cib_op_attr_transaction,
	- cib_server_process_diff
	- },
	- {
	- PCMK__CIB_REQUEST_REPLACE, cib__op_replace,
	- cib_op_attr_modifies
	- \|cib_op_attr_privileged
	- \|cib_op_attr_replaces
	- \|cib_op_attr_writes_through
	- \|cib_op_attr_transaction,
	- cib_process_replace_svr
	- },
	- {
	- PCMK__CIB_REQUEST_CREATE, cib__op_create,
	- cib_op_attr_modifies\|cib_op_attr_privileged\|cib_op_attr_transaction,
	- cib_process_create
	- },
	- {
	- PCMK__CIB_REQUEST_DELETE, cib__op_delete,
	- cib_op_attr_modifies\|cib_op_attr_privileged\|cib_op_attr_transaction,
	- cib_process_delete
	- },
	- {
	- PCMK__CIB_REQUEST_SYNC_TO_ALL, cib__op_sync_all,
	- cib_op_attr_privileged,
	- cib_process_sync
	- },
	- {
	- PCMK__CIB_REQUEST_BUMP, cib__op_bump,
	- cib_op_attr_modifies\|cib_op_attr_privileged\|cib_op_attr_transaction,
	- cib_process_bump
	- },
	- {
	- PCMK__CIB_REQUEST_ERASE, cib__op_erase,
	- cib_op_attr_modifies
	- \|cib_op_attr_privileged
	- \|cib_op_attr_replaces
	- \|cib_op_attr_transaction,
	- cib_process_erase
	- },
	- {
	- PCMK__CIB_REQUEST_NOOP, cib__op_noop,
	- cib_op_attr_none,
	- cib_process_noop
	- },
	- {
	- PCMK__CIB_REQUEST_ABS_DELETE, cib__op_abs_delete,
	- cib_op_attr_modifies\|cib_op_attr_privileged,
	- cib_process_delete_absolute
	- },
	- {
	- PCMK__CIB_REQUEST_UPGRADE, cib__op_upgrade,
	- cib_op_attr_modifies
	- \|cib_op_attr_privileged
	- \|cib_op_attr_writes_through
	- \|cib_op_attr_transaction,
	- cib_process_upgrade_server
	- },
	- {
	- PCMK__CIB_REQUEST_SECONDARY, cib__op_secondary,
	- cib_op_attr_privileged\|cib_op_attr_local,
	- cib_process_readwrite
	- },
	- {
	- PCMK__CIB_REQUEST_SYNC_TO_ONE, cib__op_sync_one,
	- cib_op_attr_privileged,
	- cib_process_sync_one
	- },
	- {
	- // @COMPAT: Drop cib_op_attr_modifies when we drop legacy mode support
	- PCMK__CIB_REQUEST_PRIMARY, cib__op_primary,
	- cib_op_attr_modifies\|cib_op_attr_privileged\|cib_op_attr_local,
	- cib_process_readwrite
	- },
	- {
	- PCMK__CIB_REQUEST_IS_PRIMARY, cib__op_is_primary,
	- cib_op_attr_privileged,
	- cib_process_readwrite
	- },
	- {
	- PCMK__CIB_REQUEST_SHUTDOWN, cib__op_shutdown,
	- cib_op_attr_privileged,
	- cib_process_shutdown_req
	- },
	- {
	- CRM_OP_PING, cib__op_ping,
	- cib_op_attr_none,
	- cib_process_ping
	- },
	+static const cib__op_fn_t cib_op_functions[] = {
	+ [cib__op_abs_delete] = cib_process_delete_absolute,
	+ [cib__op_apply_patch] = cib_server_process_diff,
	+ [cib__op_bump] = cib_process_bump,
	+ [cib__op_create] = cib_process_create,
	+ [cib__op_delete] = cib_process_delete,
	+ [cib__op_erase] = cib_process_erase,
	+ [cib__op_is_primary] = cib_process_readwrite,
	+ [cib__op_modify] = cib_process_modify,
	+ [cib__op_noop] = cib_process_noop,
	+ [cib__op_ping] = cib_process_ping,
	+ [cib__op_primary] = cib_process_readwrite,
	+ [cib__op_query] = cib_process_query,
	+ [cib__op_replace] = cib_process_replace_svr,
	+ [cib__op_secondary] = cib_process_readwrite,
	+ [cib__op_shutdown] = cib_process_shutdown_req,
	+ [cib__op_sync_all] = cib_process_sync,
	+ [cib__op_sync_one] = cib_process_sync_one,
	+ [cib__op_upgrade] = cib_process_upgrade_server,

	/* PCMK__CIB_REQUEST_*_TRANSACT requests must be processed locally because
	* they depend on the client table. Requests that manage transactions on
	* other nodes would likely be problematic in many other ways as well.
	*/
	- {
	- PCMK__CIB_REQUEST_INIT_TRANSACT, cib__op_init_transact,
	- cib_op_attr_privileged\|cib_op_attr_local,
	- cib_process_init_transaction,
	- },
	- {
	- PCMK__CIB_REQUEST_COMMIT_TRANSACT, cib__op_commit_transact,
	- cib_op_attr_modifies
	- \|cib_op_attr_privileged
	- \|cib_op_attr_local
	- \|cib_op_attr_replaces
	- \|cib_op_attr_writes_through,
	- cib_process_commit_transaction,
	- },
	- {
	- PCMK__CIB_REQUEST_DISCARD_TRANSACT, cib__op_discard_transact,
	- cib_op_attr_privileged\|cib_op_attr_local,
	- cib_process_discard_transaction,
	- },
	+ [cib__op_init_transact] = cib_process_init_transaction,
	+ [cib__op_commit_transact] = cib_process_commit_transaction,
	+ [cib__op_discard_transact] = cib_process_discard_transaction,
	};

	-int
	-cib_get_operation(const char op, const cib_operation_t *operation)
	+/*!
	+ * \internal
	+ * \brief Get the function that performs a given server-side CIB operation
	+ *
	+ * \param[in] operation Operation whose function to look up
	+ *
	+ * \return Function that performs \p operation within \c pacemaker-based
	+ */
	+cib__op_fn_t
	+based_get_op_function(const cib_operation_t *operation)
	{
	- static GHashTable *operation_hash = NULL;
	-
	- CRM_ASSERT((op != NULL) && (operation != NULL));
	-
	- if (operation_hash == NULL) {
	- operation_hash = pcmk__strkey_table(NULL, NULL);
	-
	- for (int lpc = 0; lpc < PCMK__NELEM(cib_server_ops); lpc++) {
	- const cib_operation_t *oper = &(cib_server_ops[lpc]);
	-
	- g_hash_table_insert(operation_hash, (gpointer) oper->name,
	- (gpointer) oper);
	- }
	- }
	+ enum cib__op_type type = operation->type;

	- *operation = g_hash_table_lookup(operation_hash, op);
	+ CRM_ASSERT(type >= 0);

	- if (*operation == NULL) {
	- crm_err("Operation %s is not valid", op);
	- return -EINVAL;
	+ if (type >= PCMK__NELEM(cib_op_functions)) {
	+ return NULL;
	}
	- return pcmk_ok;
	+ return cib_op_functions[type];
	}
	diff --git a/daemons/based/based_transaction.c b/daemons/based/based_transaction.c
	index d90d1b4420..d9234dd8db 100644
	--- a/daemons/based/based_transaction.c
	+++ b/daemons/based/based_transaction.c
	@@ -1,371 +1,370 @@
	/*
	* Copyright 2023 the Pacemaker project contributors
	*
	* The version control history for this file may have further details.
	*
	* This source code is licensed under the GNU General Public License version 2
	* or later (GPLv2+) WITHOUT ANY WARRANTY.
	*/

	#include <crm_internal.h>

	#include <glib.h>
	#include <libxml/tree.h>

	#include "pacemaker-based.h"

	struct request_data {
	pcmk__client_t *client;
	xmlNodePtr request;
	bool privileged;
	};

	/* Table of uncommitted CIB transactions
	* key: client ID (const char *), value: GQueue of struct request_data
	*/
	static GHashTable *transactions = NULL;

	/*!
	* \internal
	* \brief Create a new a CIB request data object
	*
	* \param[in] client CIB client
	* \param[in] request CIB request
	* \param[in] privileged If \p true, the request has write privileges
	*/
	static struct request_data *
	create_request_data(pcmk__client_t *client, xmlNodePtr request,
	bool privileged)
	{
	struct request_data *data = calloc(1, sizeof(struct request_data));

	if (data == NULL) {
	return NULL;
	}

	/* Caller owns client and request. The client's transaction is always
	* discarded before the client is freed, so we can safely use it when
	* processing requests in the transaction. The request is freed before we
	* use it, so make a copy.
	*/
	data->client = client;
	data->request = copy_xml(request);
	data->privileged = privileged;

	return data;
	}

	/*!
	* \internal
	* \brief Free a CIB request data object
	*
	* \param[in,out] data Request data object to free
	*/
	static void
	free_request_data(gpointer data)
	{
	struct request_data req_data = (struct request_data ) data;

	if (req_data != NULL) {
	// req_data doesn't own req_data->client
	free_xml(req_data->request);
	free(req_data);
	}
	}

	/*!
	* \internal
	* \brief Free a CIB transaction and the requests it contains
	*
	* \param[in,out] data Transaction to free
	*/
	static inline void
	free_transaction(gpointer data)
	{
	g_queue_free_full((GQueue *) data, (GDestroyNotify) free_request_data);
	}

	/*!
	* \internal
	* \brief Get a CIB client's uncommitted transaction, if any
	*
	* \param[in] client Client to look up
	*
	* \return The client's uncommitted transaction, if any
	*
	* \note The caller must not free the return value directly. Transactions must
	* be freed by \p based_discard_transaction() or by
	* \p based_free_transaction_table().
	*/
	static inline GQueue *
	get_transaction(const pcmk__client_t *client)
	{
	if (transactions == NULL) {
	return NULL;
	}
	return g_hash_table_lookup(transactions, client->id);
	}

	/*!
	* \internal
	* \brief Create a new transaction for a given CIB client
	*
	* \param[in] client Client to initiate a transaction for
	*
	* \return Standard Pacemaker return code
	*/
	int
	based_init_transaction(const pcmk__client_t *client)
	{
	CRM_ASSERT(client != NULL);

	if (client->id == NULL) {
	crm_warn("Can't initiate transaction for client without id");
	return EINVAL;
	}

	// A client can have at most one transaction at a time
	if (get_transaction(client) != NULL) {
	return pcmk_rc_already;
	}

	crm_trace("Initiating transaction for client %s (%s)",
	pcmk__client_name(client), client->id);

	if (transactions == NULL) {
	transactions = pcmk__strkey_table(NULL, free_transaction);
	}
	g_hash_table_insert(transactions, client->id, g_queue_new());
	return pcmk_rc_ok;
	}

	/*!
	* \internal
	* \brief Validate that a CIB request is supported in a transaction
	*
	* \param[in] request CIB request
	*
	* \return Standard Pacemaker return code
	*/
	static int
	validate_transaction_request(const xmlNode *request)
	{
	const char *op = crm_element_value(request, F_CIB_OPERATION);
	const char *host = crm_element_value(request, F_CIB_HOST);

	const cib_operation_t *operation = NULL;
	- int rc = cib_get_operation(op, &operation);
	+ int rc = cib__get_operation(op, &operation);

	- rc = pcmk_legacy2rc(rc);
	if (rc != pcmk_rc_ok) {
	// cib_get_operation() logs error
	return rc;
	}

	if (!pcmk_is_set(operation->flags, cib_op_attr_transaction)) {
	crm_err("Operation '%s' is not supported in CIB transaction", op);
	return EOPNOTSUPP;
	}

	if (!pcmk__str_eq(host, OUR_NODENAME,
	pcmk__str_casei\|pcmk__str_null_matches)) {
	crm_err("Operation targeting another node (%s) is not supported in CIB "
	"transaction",
	host);
	return EOPNOTSUPP;
	}

	return pcmk_rc_ok;
	}

	/*!
	* \internal
	* \brief Add a new CIB request to an existing transaction
	*
	* \param[in] client Client whose transaction to extend
	* \param[in] request CIB request
	* \param[in] privileged If \p true, the request has write privileges
	*
	* \return Standard Pacemaker return code
	*/
	int
	based_extend_transaction(pcmk__client_t *client, xmlNodePtr request,
	bool privileged)
	{
	struct request_data *data = NULL;
	GQueue *transaction = NULL;
	int rc = pcmk_rc_ok;

	CRM_ASSERT((client != NULL) && (request != NULL));

	transaction = get_transaction(client);
	if (transaction == NULL) {
	return pcmk_rc_no_transaction;
	}

	rc = validate_transaction_request(request);
	if (rc != pcmk_rc_ok) {
	return rc;
	}

	data = create_request_data(client, request, privileged);
	if (data == NULL) {
	return ENOMEM;
	}

	g_queue_push_tail(transaction, data);
	return pcmk_rc_ok;
	}

	/*!
	* \internal
	* \brief Free a CIB client's transaction (if any) and its requests
	*
	* \param[in] client Client whose transaction to discard
	*/
	void
	based_discard_transaction(const pcmk__client_t *client)
	{
	bool found = false;

	CRM_ASSERT(client != NULL);

	if (transactions != NULL) {
	found = g_hash_table_remove(transactions, client->id);
	}

	crm_trace("%s for client %s (%s)",
	(found? "Found and removed transaction" : "No transaction found"),
	pcmk__client_name(client),
	pcmk__s(client->id, "unidentified"));
	}

	/*!
	* \internal
	* \brief Process requests in a transaction
	*
	* Stop when a request fails or when all requests have been processed.
	*
	* \param[in,out] transaction Transaction to process
	* \param[in] client_name Client name (for logging only)
	* \param[in] client_id Client ID (for logging only)
	*
	* \return Standard Pacemaker return code
	*/
	static int
	process_transaction_requests(GQueue transaction, const char client_name,
	const char *client_id)
	{
	for (struct request_data *data = g_queue_pop_head(transaction);
	data != NULL; data = g_queue_pop_head(transaction)) {

	const char *op = crm_element_value(data->request, F_CIB_OPERATION);

	int rc = cib_process_request(data->request, data->privileged,
	data->client);

	rc = pcmk_legacy2rc(rc);
	if (rc != pcmk_rc_ok) {
	crm_err("Aborting CIB transaction for client %s (%s) due to failed "
	"%s request: %s",
	client_name, client_id, op, pcmk_rc_str(rc));
	crm_log_xml_info(data->request, "Failed request");
	free_request_data(data);
	return rc;
	}

	crm_trace("Applied %s request to transaction working CIB for client %s "
	"(%s)",
	op, client_name, client_id);
	crm_log_xml_trace(data->request, "Successful request");
	free_request_data(data);
	}

	return pcmk_rc_ok;
	}

	/*!
	* \internal
	* \brief Commit a given CIB client's transaction to a working CIB copy
	*
	* \param[in] client Client whose transaction to commit
	* \param[in,out] result_cib Where to store result CIB
	*
	* \return Standard Pacemaker return code
	*
	* \note This function is expected to be called only by
	* \p cib_process_commit_transaction().
	* \note \p result_cib is expected to be a copy of the current CIB as created by
	* \p cib_perform_op().
	* \note The caller is responsible for activating and syncing \p result_cib on
	* success, and for freeing it on failure.
	*/
	int
	based_commit_transaction(const pcmk__client_t client, xmlNodePtr result_cib)
	{
	GQueue *transaction = NULL;
	const char *client_name = NULL;
	const char *client_id = NULL;
	xmlNodePtr saved_cib = the_cib;
	int rc = pcmk_rc_ok;

	CRM_ASSERT((client != NULL) && (result_cib != NULL));

	/* *result_cib should be a copy of the_cib (created by cib_perform_op()). If
	* not, make a copy now. Change tracking isn't strictly required here
	* because:
	* * Each request in the transaction will have changes tracked and ACLs
	* checked if appropriate.
	* * cib_perform_op() will infer changes for the commit request at the end.
	*/
	CRM_CHECK((result_cib != NULL) && (result_cib != the_cib),
	*result_cib = copy_xml(the_cib));

	transaction = get_transaction(client);
	if (transaction == NULL) {
	return pcmk_rc_no_transaction;
	}

	client_name = pcmk__client_name(client);
	client_id = pcmk__s(client->id, "unidentified");

	crm_trace("Committing transaction for client %s (%s) to working CIB",
	client_name, client_id);

	// Apply all changes to a working copy of the CIB
	the_cib = *result_cib;

	rc = process_transaction_requests(transaction, client_name, client_id);

	crm_trace("Transaction commit %s for client %s (%s); discarding queue",
	((rc != pcmk_rc_ok)? "succeeded" : "failed"), client_name,
	client_id);

	// Free the transaction and (if aborted) free any remaining requests
	based_discard_transaction(client);

	/* Some request types (for example, erase) may have freed the_cib (the
	* working copy) and pointed it at a new XML object. In that case, it
	* follows that *result_cib (the working copy) was freed.
	*
	* Point *result_cib at the updated working copy stored in the_cib.
	*/
	*result_cib = the_cib;

	// Point the_cib back to the unchanged original copy
	the_cib = saved_cib;

	return rc;
	}

	/*!
	* \internal
	* \brief Free the transaction table and any uncommitted transactions
	*/
	void
	based_free_transaction_table(void)
	{
	if (transactions != NULL) {
	g_hash_table_destroy(transactions);
	}
	}
	diff --git a/daemons/based/pacemaker-based.h b/daemons/based/pacemaker-based.h
	index a7eafebf36..128a202a50 100644
	--- a/daemons/based/pacemaker-based.h
	+++ b/daemons/based/pacemaker-based.h
	@@ -1,153 +1,153 @@
	/*
	* Copyright 2004-2023 the Pacemaker project contributors
	*
	* The version control history for this file may have further details.
	*
	* This source code is licensed under the GNU Lesser General Public License
	* version 2.1 or later (LGPLv2.1+) WITHOUT ANY WARRANTY.
	*/

	#ifndef PACEMAKER_BASED__H
	# define PACEMAKER_BASED__H

	#include <stdio.h>
	#include <sys/types.h>
	#include <unistd.h>
	#include <stdlib.h>
	#include <glib.h>
	#include <errno.h>
	#include <fcntl.h>

	#include <glib.h>
	#include <libxml/tree.h>

	#include <crm/crm.h>
	#include <crm/cib.h>
	#include <crm/common/xml.h>
	#include <crm/cluster.h>
	#include <crm/common/ipc_internal.h>
	#include <crm/common/mainloop.h>
	#include <crm/cib/internal.h>

	#include "based_transaction.h"

	#ifdef HAVE_GNUTLS_GNUTLS_H
	# include <gnutls/gnutls.h>
	#endif

	#define OUR_NODENAME (stand_alone? "localhost" : crm_cluster->uname)

	// CIB-specific client flags
	enum cib_client_flags {
	// Notifications
	cib_notify_pre = (UINT64_C(1) << 0),
	cib_notify_post = (UINT64_C(1) << 1),
	cib_notify_replace = (UINT64_C(1) << 2),
	cib_notify_confirm = (UINT64_C(1) << 3),
	cib_notify_diff = (UINT64_C(1) << 4),

	// Whether client is another cluster daemon
	cib_is_daemon = (UINT64_C(1) << 12),
	};

	extern bool based_is_primary;
	extern GHashTable *config_hash;
	extern xmlNode *the_cib;
	extern crm_trigger_t *cib_writer;
	extern gboolean cib_writes_enabled;

	extern GMainLoop *mainloop;
	extern crm_cluster_t *crm_cluster;
	extern GHashTable *local_notify_queue;
	extern gboolean legacy_mode;
	extern gboolean stand_alone;
	extern gboolean cib_shutdown_flag;
	extern gchar *cib_root;
	extern int cib_status;
	extern pcmk__output_t *logger_out;

	extern struct qb_ipcs_service_handlers ipc_ro_callbacks;
	extern struct qb_ipcs_service_handlers ipc_rw_callbacks;
	extern qb_ipcs_service_t *ipcs_ro;
	extern qb_ipcs_service_t *ipcs_rw;
	extern qb_ipcs_service_t *ipcs_shm;

	void cib_peer_callback(xmlNode msg, void private_data);
	void cib_common_callback_worker(uint32_t id, uint32_t flags,
	xmlNode op_request, pcmk__client_t cib_client,
	gboolean privileged);
	int cib_process_request(xmlNode *request, gboolean privileged,
	const pcmk__client_t *cib_client);
	void cib_shutdown(int nsig);
	void terminate_cib(const char *caller, int fast);
	gboolean cib_legacy_mode(void);

	gboolean uninitializeCib(void);
	xmlNode readCibXmlFile(const char dir, const char *file,
	gboolean discard_status);
	int activateCibXml(xmlNode doc, gboolean to_disk, const char op);

	int cib_process_shutdown_req(const char op, int options, const char section,
	xmlNode req, xmlNode input,
	xmlNode existing_cib, xmlNode *result_cib,
	xmlNode **answer);
	int cib_process_noop(const char op, int options, const char section,
	xmlNode req, xmlNode input, xmlNode *existing_cib,
	xmlNode result_cib, xmlNode answer);
	int cib_process_ping(const char op, int options, const char section,
	xmlNode req, xmlNode input, xmlNode *existing_cib,
	xmlNode result_cib, xmlNode answer);
	int cib_process_readwrite(const char op, int options, const char section,
	xmlNode req, xmlNode input, xmlNode *existing_cib,
	xmlNode result_cib, xmlNode answer);
	int cib_process_replace_svr(const char op, int options, const char section,
	xmlNode req, xmlNode input, xmlNode *existing_cib,
	xmlNode result_cib, xmlNode answer);
	int cib_server_process_diff(const char op, int options, const char section,
	xmlNode req, xmlNode input, xmlNode *existing_cib,
	xmlNode result_cib, xmlNode answer);
	int cib_process_sync(const char op, int options, const char section,
	xmlNode req, xmlNode input, xmlNode *existing_cib,
	xmlNode result_cib, xmlNode answer);
	int cib_process_sync_one(const char op, int options, const char section,
	xmlNode req, xmlNode input, xmlNode *existing_cib,
	xmlNode result_cib, xmlNode answer);
	int cib_process_delete_absolute(const char *op, int options,
	const char section, xmlNode req,
	xmlNode input, xmlNode existing_cib,
	xmlNode result_cib, xmlNode answer);
	int cib_process_upgrade_server(const char op, int options, const char section,
	xmlNode req, xmlNode input,
	xmlNode existing_cib, xmlNode *result_cib,
	xmlNode **answer);
	int cib_process_init_transaction(const char *op, int options,
	const char section, xmlNode req,
	xmlNode input, xmlNode existing_cib,
	xmlNode result_cib, xmlNode answer);
	int cib_process_commit_transaction(const char *op, int options,
	const char section, xmlNode req,
	xmlNode input, xmlNode existing_cib,
	xmlNode result_cib, xmlNode answer);
	int cib_process_discard_transaction(const char *op, int options,
	const char section, xmlNode req,
	xmlNode input, xmlNode existing_cib,
	xmlNode result_cib, xmlNode answer);
	void send_sync_request(const char *host);
	int sync_our_cib(xmlNode *request, gboolean all);

	-int cib_get_operation(const char op, const cib_operation_t *operation);
	+cib__op_fn_t based_get_op_function(const cib_operation_t *operation);
	void cib_diff_notify(const char op, int result, const char call_id,
	const char client_id, const char client_name,
	const char origin, xmlNode update, xmlNode *diff);
	void cib_replace_notify(const char op, int result, const char call_id,
	const char client_id, const char client_name,
	const char origin, xmlNode update, xmlNode *diff,
	uint32_t change_section);

	static inline const char *
	cib_config_lookup(const char *opt)
	{
	return g_hash_table_lookup(config_hash, opt);
	}

	#endif // PACEMAKER_BASED__H
	diff --git a/include/crm/cib/internal.h b/include/crm/cib/internal.h
	index 35aff95fa0..c8649bdb52 100644
	--- a/include/crm/cib/internal.h
	+++ b/include/crm/cib/internal.h
	@@ -1,332 +1,333 @@
	/*
	* Copyright 2004-2023 the Pacemaker project contributors
	*
	* The version control history for this file may have further details.
	*
	* This source code is licensed under the GNU Lesser General Public License
	* version 2.1 or later (LGPLv2.1+) WITHOUT ANY WARRANTY.
	*/

	#ifndef CIB_INTERNAL__H
	# define CIB_INTERNAL__H
	# include <crm/cib.h>
	# include <crm/common/ipc_internal.h>
	# include <crm/common/output_internal.h>

	// Request types for CIB manager IPC/CPG
	#define PCMK__CIB_REQUEST_SECONDARY "cib_slave"
	#define PCMK__CIB_REQUEST_PRIMARY "cib_master"
	#define PCMK__CIB_REQUEST_SYNC_TO_ALL "cib_sync"
	#define PCMK__CIB_REQUEST_SYNC_TO_ONE "cib_sync_one"
	#define PCMK__CIB_REQUEST_IS_PRIMARY "cib_ismaster"
	#define PCMK__CIB_REQUEST_BUMP "cib_bump"
	#define PCMK__CIB_REQUEST_QUERY "cib_query"
	#define PCMK__CIB_REQUEST_CREATE "cib_create"
	#define PCMK__CIB_REQUEST_MODIFY "cib_modify"
	#define PCMK__CIB_REQUEST_DELETE "cib_delete"
	#define PCMK__CIB_REQUEST_ERASE "cib_erase"
	#define PCMK__CIB_REQUEST_REPLACE "cib_replace"
	#define PCMK__CIB_REQUEST_APPLY_PATCH "cib_apply_diff"
	#define PCMK__CIB_REQUEST_UPGRADE "cib_upgrade"
	#define PCMK__CIB_REQUEST_ABS_DELETE "cib_delete_alt"
	#define PCMK__CIB_REQUEST_NOOP "noop"
	#define PCMK__CIB_REQUEST_SHUTDOWN "cib_shutdown_req"
	#define PCMK__CIB_REQUEST_INIT_TRANSACT "cib_init_transact"
	#define PCMK__CIB_REQUEST_COMMIT_TRANSACT "cib_commit_transact"
	#define PCMK__CIB_REQUEST_DISCARD_TRANSACT "cib_discard_transact"

	# define F_CIB_CLIENTID "cib_clientid"
	# define F_CIB_CALLOPTS "cib_callopt"
	# define F_CIB_CALLID "cib_callid"
	# define F_CIB_CALLDATA "cib_calldata"
	# define F_CIB_OPERATION "cib_op"
	# define F_CIB_ISREPLY "cib_isreplyto"
	# define F_CIB_SECTION "cib_section"
	# define F_CIB_HOST "cib_host"
	# define F_CIB_RC "cib_rc"
	# define F_CIB_UPGRADE_RC "cib_upgrade_rc"
	# define F_CIB_DELEGATED "cib_delegated_from"
	# define F_CIB_OBJID "cib_object"
	# define F_CIB_OBJTYPE "cib_object_type"
	# define F_CIB_EXISTING "cib_existing_object"
	# define F_CIB_SEENCOUNT "cib_seen"
	# define F_CIB_TIMEOUT "cib_timeout"
	# define F_CIB_UPDATE "cib_update"
	# define F_CIB_GLOBAL_UPDATE "cib_update"
	# define F_CIB_UPDATE_RESULT "cib_update_result"
	# define F_CIB_CLIENTNAME "cib_clientname"
	# define F_CIB_NOTIFY_TYPE "cib_notify_type"
	# define F_CIB_NOTIFY_ACTIVATE "cib_notify_activate"
	# define F_CIB_UPDATE_DIFF "cib_update_diff"
	# define F_CIB_USER "cib_user"
	# define F_CIB_LOCAL_NOTIFY_ID "cib_local_notify_id"
	# define F_CIB_PING_ID "cib_ping_id"
	# define F_CIB_SCHEMA_MAX "cib_schema_max"
	# define F_CIB_CHANGE_SECTION "cib_change_section"

	# define T_CIB "cib"
	# define T_CIB_NOTIFY "cib_notify"
	/* notify sub-types */
	# define T_CIB_PRE_NOTIFY "cib_pre_notify"
	# define T_CIB_POST_NOTIFY "cib_post_notify"
	# define T_CIB_UPDATE_CONFIRM "cib_update_confirmation"
	# define T_CIB_REPLACE_NOTIFY "cib_refresh_notify"

	/*!
	* \internal
	* \enum cib_change_section_info
	* \brief Flags to indicate which sections of the CIB have changed
	*/
	enum cib_change_section_info {
	cib_change_section_none = 0, //!< No sections have changed
	cib_change_section_nodes = (1 << 0), //!< The nodes section has changed
	cib_change_section_alerts = (1 << 1), //!< The alerts section has changed
	cib_change_section_status = (1 << 2), //!< The status section has changed
	};

	/*!
	* \internal
	* \enum cib_op_attr
	* \brief Flags for CIB operation attributes
	*/
	enum cib_op_attr {
	cib_op_attr_none = 0, //!< No special attributes
	cib_op_attr_modifies = (1 << 1), //!< Modifies CIB
	cib_op_attr_privileged = (1 << 2), //!< Requires privileges
	cib_op_attr_local = (1 << 3), //!< Must only be processed locally
	cib_op_attr_replaces = (1 << 4), //!< Replaces CIB
	cib_op_attr_writes_through = (1 << 5), //!< Writes to disk on success
	cib_op_attr_transaction = (1 << 6), //!< Supported in a transaction
	};

	/*!
	* \internal
	* \enum cib__op_type
	* \brief Types of CIB operations
	*/
	enum cib__op_type {
	cib__op_abs_delete,
	cib__op_apply_patch,
	cib__op_bump,
	cib__op_create,
	cib__op_delete,
	cib__op_erase,
	cib__op_is_primary,
	cib__op_modify,
	cib__op_noop,
	cib__op_ping,
	cib__op_primary,
	cib__op_query,
	cib__op_replace,
	cib__op_secondary,
	cib__op_shutdown,
	cib__op_sync_all,
	cib__op_sync_one,
	cib__op_upgrade,

	// @TODO: Refactor transactions and remove these
	cib__op_init_transact,
	cib__op_commit_transact,
	cib__op_discard_transact,
	};

	/*!
	* \internal
	* \brief Set given <tt>enum cib_change_section_info</tt> flags
	*
	* \param[in,out] flags_orig Group of flags to update
	* \param[in] flags_to_set Flags to clear from \p flags_orig
	*/
	#define pcmk__set_change_section(flags_orig, flags_to_set) do { \
	flags_orig = pcmk__set_flags_as(__func__, __LINE__, LOG_TRACE, \
	"CIB change section", \
	"change_section", flags_orig, \
	flags_to_set, #flags_to_set); \
	} while (0)

	gboolean cib_diff_version_details(xmlNode * diff, int admin_epoch, int epoch, int *updates,
	int _admin_epoch, int _epoch, int *_updates);

	gboolean cib_read_config(GHashTable * options, xmlNode * current_cib);

	typedef int (cib__op_fn_t)(const char , int, const char , xmlNode ,
	xmlNode , xmlNode , xmlNode , xmlNode );

	typedef struct cib_operation_s {
	const char *name;
	enum cib__op_type type;
	uint32_t flags; //!< Group of <tt>enum cib_op_attr</tt> flags
	- cib__op_fn_t fn;
	} cib_operation_t;

	typedef struct cib_notify_client_s {
	const char *event;
	const char obj_id; / implement one day */
	const char obj_type; / implement one day */
	void (callback) (const char event, xmlNode * msg);

	} cib_notify_client_t;

	typedef struct cib_callback_client_s {
	void (callback) (xmlNode , int, int, xmlNode , void );
	const char *id;
	void *user_data;
	gboolean only_success;
	struct timer_rec_s *timer;
	void (free_func)(void );
	} cib_callback_client_t;

	struct timer_rec_s {
	int call_id;
	int timeout;
	guint ref;
	cib_t *cib;
	};

	#define cib__set_call_options(cib_call_opts, call_for, flags_to_set) do { \
	cib_call_opts = pcmk__set_flags_as(__func__, __LINE__, \
	LOG_TRACE, "CIB call", (call_for), (cib_call_opts), \
	(flags_to_set), #flags_to_set); \
	} while (0)

	#define cib__clear_call_options(cib_call_opts, call_for, flags_to_clear) do { \
	cib_call_opts = pcmk__clear_flags_as(__func__, __LINE__, \
	LOG_TRACE, "CIB call", (call_for), (cib_call_opts), \
	(flags_to_clear), #flags_to_clear); \
	} while (0)

	cib_t *cib_new_variant(void);

	int cib_perform_op(const char *op, int call_options, cib__op_fn_t fn,
	bool is_query, const char section, xmlNode req,
	xmlNode input, bool manage_counters, bool config_changed,
	xmlNode current_cib, xmlNode result_cib, xmlNode **diff,
	xmlNode **output);

	xmlNode cib_create_op(int call_id, const char op, const char *host,
	const char section, xmlNode data, int call_options,
	const char *user_name);

	void cib_native_callback(cib_t * cib, xmlNode * msg, int call_id, int rc);
	void cib_native_notify(gpointer data, gpointer user_data);

	+int cib__get_operation(const char op, const cib_operation_t *operation);
	+
	int cib_process_query(const char op, int options, const char section, xmlNode * req,
	xmlNode * input, xmlNode * existing_cib, xmlNode ** result_cib,
	xmlNode ** answer);

	int cib_process_erase(const char op, int options, const char section, xmlNode * req,
	xmlNode * input, xmlNode * existing_cib, xmlNode ** result_cib,
	xmlNode ** answer);

	int cib_process_bump(const char op, int options, const char section, xmlNode * req,
	xmlNode * input, xmlNode * existing_cib, xmlNode ** result_cib,
	xmlNode ** answer);

	int cib_process_replace(const char op, int options, const char section, xmlNode * req,
	xmlNode * input, xmlNode * existing_cib, xmlNode ** result_cib,
	xmlNode ** answer);

	int cib_process_create(const char op, int options, const char section, xmlNode * req,
	xmlNode * input, xmlNode * existing_cib, xmlNode ** result_cib,
	xmlNode ** answer);

	int cib_process_modify(const char op, int options, const char section, xmlNode * req,
	xmlNode * input, xmlNode * existing_cib, xmlNode ** result_cib,
	xmlNode ** answer);

	int cib_process_delete(const char op, int options, const char section, xmlNode * req,
	xmlNode * input, xmlNode * existing_cib, xmlNode ** result_cib,
	xmlNode ** answer);

	int cib_process_diff(const char op, int options, const char section, xmlNode * req,
	xmlNode * input, xmlNode * existing_cib, xmlNode ** result_cib,
	xmlNode ** answer);

	int cib_process_upgrade(const char op, int options, const char section, xmlNode * req,
	xmlNode * input, xmlNode * existing_cib, xmlNode ** result_cib,
	xmlNode ** answer);

	/*!
	* \internal
	* \brief Query or modify a CIB
	*
	* \param[in] op PCMK__CIB_REQUEST_* operation to be performed
	* \param[in] options Flag set of \c cib_call_options
	* \param[in] section XPath to query or modify
	* \param[in] req unused
	* \param[in] input Portion of CIB to modify (used with
	* PCMK__CIB_REQUEST_CREATE,
	* PCMK__CIB_REQUEST_MODIFY, and
	* PCMK__CIB_REQUEST_REPLACE)
	* \param[in,out] existing_cib Input CIB (used with PCMK__CIB_REQUEST_QUERY)
	* \param[in,out] result_cib CIB copy to make changes in (used with
	* PCMK__CIB_REQUEST_CREATE,
	* PCMK__CIB_REQUEST_MODIFY,
	* PCMK__CIB_REQUEST_DELETE, and
	* PCMK__CIB_REQUEST_REPLACE)
	* \param[out] answer Query result (used with PCMK__CIB_REQUEST_QUERY)
	*
	* \return Legacy Pacemaker return code
	*/
	int cib_process_xpath(const char op, int options, const char section,
	const xmlNode req, xmlNode input, xmlNode *existing_cib,
	xmlNode result_cib, xmlNode answer);

	bool cib__config_changed_v1(xmlNode last, xmlNode next, xmlNode **diff);

	int cib_internal_op(cib_t * cib, const char op, const char host,
	const char section, xmlNode data,
	xmlNode ** output_data, int call_options, const char *user_name);


	int cib_file_read_and_verify(const char filename, const char sigfile,
	xmlNode **root);
	int cib_file_write_with_digest(xmlNode cib_root, const char cib_dirname,
	const char *cib_filename);

	void cib__set_output(cib_t cib, pcmk__output_t out);

	cib_callback_client_t* cib__lookup_id (int call_id);

	/*!
	* \internal
	* \brief Connect to, query, and optionally disconnect from the CIB
	*
	* Open a read-write connection to the CIB manager if an already connected
	* client is not passed in. Then query the CIB and store the resulting XML.
	* Finally, disconnect if the CIB connection isn't being returned to the caller.
	*
	* \param[in,out] out Output object (may be \p NULL)
	* \param[in,out] cib If not \p NULL, where to store CIB connection
	* \param[out] cib_object Where to store query result
	*
	* \return Standard Pacemaker return code
	*
	* \note If \p cib is not \p NULL, the caller is responsible for freeing \p *cib
	* using \p cib_delete().
	* \note If \p *cib points to an existing \p cib_t object, this function will
	* reuse it instead of creating a new one. If the existing client is
	* already connected, the connection will be reused, even if it's
	* read-only.
	*/
	int cib__signon_query(pcmk__output_t out, cib_t cib, xmlNode *cib_object);

	int cib__clean_up_connection(cib_t **cib);

	int cib__update_node_attr(pcmk__output_t out, cib_t cib, int call_options,
	const char section, const char node_uuid, const char *set_type,
	const char set_name, const char attr_id, const char *attr_name,
	const char attr_value, const char user_name,
	const char *node_type);

	int cib__get_node_attrs(pcmk__output_t out, cib_t cib, const char *section,
	const char node_uuid, const char set_type, const char *set_name,
	const char attr_id, const char attr_name, const char *user_name,
	xmlNode **result);

	int cib__delete_node_attr(pcmk__output_t out, cib_t cib, int options,
	const char section, const char node_uuid, const char *set_type,
	const char set_name, const char attr_id, const char *attr_name,
	const char attr_value, const char user_name);

	#endif
	diff --git a/lib/cib/cib_file.c b/lib/cib/cib_file.c
	index dd1d9f3bd8..51edfb21e0 100644
	--- a/lib/cib/cib_file.c
	+++ b/lib/cib/cib_file.c
	@@ -1,1419 +1,1341 @@
	/*
	* Original copyright 2004 International Business Machines
	* Later changes copyright 2008-2023 the Pacemaker project contributors
	*
	* The version control history for this file may have further details.
	*
	* This source code is licensed under the GNU Lesser General Public License
	* version 2.1 or later (LGPLv2.1+) WITHOUT ANY WARRANTY.
	*/

	#include <crm_internal.h>
	#include <unistd.h>
	#include <limits.h>
	#include <stdlib.h>
	#include <stdint.h>
	#include <stdio.h>
	#include <stdarg.h>
	#include <string.h>
	#include <pwd.h>

	#include <sys/stat.h>
	#include <sys/types.h>
	#include <glib.h>

	#include <crm/crm.h>
	#include <crm/cib/internal.h>
	#include <crm/msg_xml.h>
	#include <crm/common/ipc.h>
	#include <crm/common/xml.h>
	#include <crm/common/xml_internal.h>

	#define CIB_SERIES "cib"
	#define CIB_SERIES_MAX 100
	#define CIB_SERIES_BZIP FALSE /* Must be false because archived copies are
	created with hard links
	*/

	#define CIB_LIVE_NAME CIB_SERIES ".xml"

	// key: client ID (const char ) -> value: client (cib_t )
	static GHashTable *client_table = NULL;

	-// key: op name (const char ) -> value: operation (const cib_operation_t )
	-static GHashTable *op_table = NULL;
	-
	enum cib_file_flags {
	cib_file_flag_dirty = (1 << 0),
	cib_file_flag_live = (1 << 1),
	};

	typedef struct cib_file_opaque_s {
	char *id;
	char *filename;
	uint32_t flags; // Group of enum cib_file_flags
	xmlNode *cib_xml;
	GQueue *transaction;
	} cib_file_opaque_t;

	static int cib_file_extend_transaction(cib_t *cib,
	const cib_operation_t *operation,
	xmlNode *request);

	static void cib_file_discard_transaction(cib_t *cib);

	static int cib_file_process_init_transaction(const char *op, int options,
	const char section, xmlNode req,
	xmlNode *input,
	xmlNode *existing_cib,
	xmlNode **result_cib,
	xmlNode **answer);

	static int cib_file_process_commit_transaction(const char *op, int options,
	const char *section,
	xmlNode req, xmlNode input,
	xmlNode *existing_cib,
	xmlNode **result_cib,
	xmlNode **answer);

	static int cib_file_process_discard_transaction(const char *op, int options,
	const char *section,
	xmlNode req, xmlNode input,
	xmlNode *existing_cib,
	xmlNode **result_cib,
	xmlNode **answer);
	/*!
	* \internal
	* \brief Add a CIB file client to client table
	*
	* \param[in] cib CIB client
	*/
	static void
	register_client(const cib_t *cib)
	{
	cib_file_opaque_t *private = cib->variant_opaque;

	if (client_table == NULL) {
	client_table = pcmk__strkey_table(NULL, NULL);
	}
	g_hash_table_insert(client_table, private->id, (gpointer) cib);
	}

	/*!
	* \internal
	* \brief Remove a CIB file client from client table
	*
	* \param[in] cib CIB client
	*/
	static void
	unregister_client(const cib_t *cib)
	{
	cib_file_opaque_t *private = cib->variant_opaque;

	if (client_table == NULL) {
	return;
	}

	g_hash_table_remove(client_table, private->id);

	- /* There's no clean way to free the tables at program exit time. We can't
	- * create a cleanup function in this file and add it to crm_exit(), which is
	- * in libcrmcommon. libcrmcommon doesn't link against libcib.
	- *
	- * As a hack for now, free the common tables when there are no more clients.
	- *
	- * @COMPAT: If libcib and libcrmcommon are merged, add this to crm_exit().
	+ /* @COMPAT: Add to crm_exit() when libcib and libcrmcommon are merged,
	+ * instead of destroying the client table when there are no more clients.
	*/
	if (g_hash_table_size(client_table) == 0) {
	g_hash_table_destroy(client_table);
	client_table = NULL;
	-
	- if (op_table != NULL) {
	- g_hash_table_destroy(op_table);
	- op_table = NULL;
	- }
	}
	}

	/*!
	* \internal
	* \brief Look up a CIB file client by its ID
	*
	* \param[in] client_id CIB client ID
	*
	* \return CIB client with matching ID if found, or \p NULL otherwise
	*/
	static cib_t *
	get_client(const char *client_id)
	{
	if (client_table == NULL) {
	return NULL;
	}
	return g_hash_table_lookup(client_table, (gpointer) client_id);
	}

	-/* This table is adapted from cib_server_ops in pacemaker-based. Flags that are
	- * not used in cib_file.c (for example, cib_op_attr_replaces) are not currently
	- * included. Also, cib_file.c does not use prepare or cleanup functions.
	- *
	- * @TODO: Try to create a shared table that merges cib_server_ops with
	- * cib_file_ops. We would want a new cib_op_attr_file enum value (for an
	- * operation that supports running against a file). We can easily ignore the
	- * prepare and cleanup functions.
	- *
	- * The main challenge is that the fn member (cib__op_fn_t) may differ depending
	- * on whether we're processing the request on the server side or with the
	- * file-based client. Processor functions defined in pacemaker-based shouldn't
	- * be declared in libcib, but we need them accessible via the table somehow.
	- *
	- * Maybe use an enum that indexes into an array of cib__op_fn_t functions. That
	- * would be obnoxiously similar to the call_type (index) argument that we
	- * eliminated in c13bf8b2.
	- */
	-static const cib_operation_t cib_file_ops[] = {
	- {
	- PCMK__CIB_REQUEST_APPLY_PATCH, cib__op_apply_patch,
	- cib_op_attr_modifies\|cib_op_attr_transaction,
	- cib_process_diff
	- },
	- {
	- PCMK__CIB_REQUEST_BUMP, cib__op_bump,
	- cib_op_attr_modifies\|cib_op_attr_transaction,
	- cib_process_bump
	- },
	- {
	- PCMK__CIB_REQUEST_CREATE, cib__op_create,
	- cib_op_attr_modifies\|cib_op_attr_transaction,
	- cib_process_create
	- },
	- {
	- PCMK__CIB_REQUEST_DELETE, cib__op_delete,
	- cib_op_attr_modifies\|cib_op_attr_transaction,
	- cib_process_delete
	- },
	- {
	- PCMK__CIB_REQUEST_ERASE, cib__op_erase,
	- cib_op_attr_modifies\|cib_op_attr_transaction,
	- cib_process_erase
	- },
	- {
	- PCMK__CIB_REQUEST_MODIFY, cib__op_modify,
	- cib_op_attr_modifies\|cib_op_attr_transaction,
	- cib_process_modify
	- },
	- {
	- PCMK__CIB_REQUEST_QUERY, cib__op_query,
	- cib_op_attr_none,
	- cib_process_query
	- },
	- {
	- PCMK__CIB_REQUEST_REPLACE, cib__op_replace,
	- cib_op_attr_modifies\|cib_op_attr_transaction,
	- cib_process_replace
	- },
	- {
	- PCMK__CIB_REQUEST_UPGRADE, cib__op_upgrade,
	- cib_op_attr_modifies\|cib_op_attr_transaction,
	- cib_process_upgrade
	- },
	- {
	- PCMK__CIB_REQUEST_INIT_TRANSACT, cib__op_init_transact,
	- cib_op_attr_none,
	- cib_file_process_init_transaction
	- },
	- {
	- PCMK__CIB_REQUEST_COMMIT_TRANSACT, cib__op_commit_transact,
	- cib_op_attr_modifies,
	- cib_file_process_commit_transaction
	- },
	- {
	- PCMK__CIB_REQUEST_DISCARD_TRANSACT, cib__op_discard_transact,
	- cib_op_attr_none,
	- cib_file_process_discard_transaction
	- },
	+static const cib__op_fn_t cib_op_functions[] = {
	+ [cib__op_apply_patch] = cib_process_diff,
	+ [cib__op_bump] = cib_process_bump,
	+ [cib__op_create] = cib_process_create,
	+ [cib__op_delete] = cib_process_delete,
	+ [cib__op_erase] = cib_process_erase,
	+ [cib__op_modify] = cib_process_modify,
	+ [cib__op_query] = cib_process_query,
	+ [cib__op_replace] = cib_process_replace,
	+ [cib__op_upgrade] = cib_process_upgrade,
	+ [cib__op_init_transact] = cib_file_process_init_transaction,
	+ [cib__op_commit_transact] = cib_file_process_commit_transaction,
	+ [cib__op_discard_transact] = cib_file_process_discard_transaction,
	};

	/* cib_file_backup() and cib_file_write_with_digest() need to chown the
	* written files only in limited circumstances, so these variables allow
	* that to be indicated without affecting external callers
	*/
	static uid_t cib_file_owner = 0;
	static uid_t cib_file_group = 0;
	static gboolean cib_do_chown = FALSE;

	#define cib_set_file_flags(cibfile, flags_to_set) do { \
	(cibfile)->flags = pcmk__set_flags_as(__func__, __LINE__, \
	LOG_TRACE, "CIB file", \
	cibfile->filename, \
	(cibfile)->flags, \
	(flags_to_set), \
	#flags_to_set); \
	} while (0)

	#define cib_clear_file_flags(cibfile, flags_to_clear) do { \
	(cibfile)->flags = pcmk__clear_flags_as(__func__, __LINE__, \
	LOG_TRACE, "CIB file", \
	cibfile->filename, \
	(cibfile)->flags, \
	(flags_to_clear), \
	#flags_to_clear); \
	} while (0)

	-static int
	-get_operation(const char op, const cib_operation_t *operation)
	+/*!
	+ * \internal
	+ * \brief Get the function that performs a given CIB file operation
	+ *
	+ * \param[in] operation Operation whose function to look up
	+ *
	+ * \return Function that performs \p operation for a CIB file client
	+ */
	+static cib__op_fn_t
	+file_get_op_function(const cib_operation_t *operation)
	{
	- CRM_CHECK(op != NULL, return -EINVAL);
	-
	- if (op_table == NULL) {
	- op_table = pcmk__strkey_table(NULL, NULL);
	+ enum cib__op_type type = operation->type;

	- for (int lpc = 0; lpc < PCMK__NELEM(cib_file_ops); lpc++) {
	- const cib_operation_t *oper = &(cib_file_ops[lpc]);
	+ CRM_ASSERT(type >= 0);

	- g_hash_table_insert(op_table, (gpointer) oper->name,
	- (gpointer) oper);
	- }
	- }
	-
	- *operation = g_hash_table_lookup(op_table, op);
	-
	- if (*operation == NULL) {
	- /* @COMPAT: EOPNOTSUPP makes more sense but would change existing
	- * behavior. If we merge the server and file ops tables in the future,
	- * then EINVAL makes sense: if we didn't find the op, it doesn't exist.
	- * Only an internal bug gets us to that point.
	- */
	- crm_err("Operation %s is invalid or unsupported for CIB file clients",
	- op);
	- return -EPROTONOSUPPORT;
	+ if (type >= PCMK__NELEM(cib_op_functions)) {
	+ return NULL;
	}
	- return pcmk_ok;
	+ return cib_op_functions[type];
	}

	/*!
	* \internal
	* \brief Check whether a file is the live CIB
	*
	* \param[in] filename Name of file to check
	*
	* \return TRUE if file exists and its real path is same as live CIB's
	*/
	static gboolean
	cib_file_is_live(const char *filename)
	{
	gboolean same = FALSE;

	if (filename != NULL) {
	// Canonicalize file names for true comparison
	char *real_filename = NULL;

	if (pcmk__real_path(filename, &real_filename) == pcmk_rc_ok) {
	char *real_livename = NULL;

	if (pcmk__real_path(CRM_CONFIG_DIR "/" CIB_LIVE_NAME,
	&real_livename) == pcmk_rc_ok) {
	same = !strcmp(real_filename, real_livename);
	free(real_livename);
	}
	free(real_filename);
	}
	}
	return same;
	}

	static int
	cib_file_process_request(cib_t cib, xmlNode request, xmlNode **output)
	{
	int rc = pcmk_ok;
	const cib_operation_t *operation = NULL;
	+ cib__op_fn_t op_function = NULL;

	int call_id = 0;
	int call_options = cib_none;
	const char *op = crm_element_value(request, F_CIB_OPERATION);
	const char *section = crm_element_value(request, F_CIB_SECTION);
	xmlNode *data = get_message_xml(request, F_CIB_CALLDATA);

	bool changed = false;
	bool read_only = false;
	xmlNode *result_cib = NULL;
	xmlNode *cib_diff = NULL;

	cib_file_opaque_t *private = cib->variant_opaque;

	- rc = get_operation(op, &operation);
	- if (rc != pcmk_ok) {
	- goto done;
	- }
	+ // We error checked these in callers
	+ cib__get_operation(op, &operation);
	+ op_function = file_get_op_function(operation);

	crm_element_value_int(request, F_CIB_CALLID, &call_id);
	crm_element_value_int(request, F_CIB_CALLOPTS, &call_options);

	read_only = !pcmk_is_set(operation->flags, cib_op_attr_modifies);

	// Mirror the logic in cib_prepare_common()
	if ((section != NULL) && (data != NULL)
	&& pcmk__str_eq(crm_element_name(data), XML_TAG_CIB, pcmk__str_none)) {

	data = pcmk_find_cib_element(data, section);
	}

	- rc = cib_perform_op(op, call_options, operation->fn, read_only, section,
	+ rc = cib_perform_op(op, call_options, op_function, read_only, section,
	request, data, true, &changed, &private->cib_xml,
	&result_cib, &cib_diff, output);

	if (pcmk_is_set(call_options, cib_transaction)) {
	goto done;
	}

	if (rc == -pcmk_err_schema_validation) {
	validate_xml_verbose(result_cib);

	} else if ((rc == pcmk_ok) && !read_only) {
	pcmk__output_t *out = NULL;

	rc = pcmk_rc2legacy(pcmk__log_output_new(&out));
	CRM_CHECK(rc == pcmk_ok, goto done);

	pcmk__output_set_log_level(out, LOG_DEBUG);
	rc = out->message(out, "xml-patchset", cib_diff);
	out->finish(out, pcmk_rc2exitc(rc), true, NULL);
	pcmk__output_free(out);
	rc = pcmk_ok;

	if (result_cib != private->cib_xml) {
	free_xml(private->cib_xml);
	private->cib_xml = result_cib;
	}
	cib_set_file_flags(private, cib_file_flag_dirty);
	}

	// Global operation callback (deprecated)
	if (cib->op_callback != NULL) {
	cib->op_callback(NULL, call_id, rc, *output);
	}

	done:
	if ((result_cib != private->cib_xml) && (result_cib != *output)) {
	free_xml(result_cib);
	}
	free_xml(cib_diff);
	return rc;
	}

	static int
	cib_file_perform_op_delegate(cib_t cib, const char op, const char *host,
	const char section, xmlNode data,
	xmlNode **output_data, int call_options,
	const char *user_name)
	{
	int rc = pcmk_ok;
	xmlNode *request = NULL;
	xmlNode *output = NULL;
	cib_file_opaque_t *private = cib->variant_opaque;

	const cib_operation_t *operation = NULL;

	crm_info("Handling %s operation for %s as %s",
	pcmk__s(op, "invalid"), pcmk__s(section, "entire CIB"),
	pcmk__s(user_name, "default user"));

	if (output_data != NULL) {
	*output_data = NULL;
	}

	if (cib->state == cib_disconnected) {
	return -ENOTCONN;
	}

	- rc = get_operation(op, &operation);
	+ rc = cib__get_operation(op, &operation);
	+ rc = pcmk_rc2legacy(rc);
	if (rc != pcmk_ok) {
	- return rc;
	+ // @COMPAT: At compatibility break, use rc directly
	+ return -EPROTONOSUPPORT;
	+ }
	+
	+ if (file_get_op_function(operation) == NULL) {
	+ // @COMPAT: At compatibility break, use EOPNOTSUPP
	+ crm_err("Operation %s is not supported by CIB file clients", op);
	+ return -EPROTONOSUPPORT;
	}

	cib->call_id++;
	cib__set_call_options(call_options, "file operation", cib_no_mtime);

	request = cib_create_op(cib->call_id, op, host, section, data, call_options,
	user_name);
	crm_xml_add(request, XML_ACL_TAG_USER, user_name);
	crm_xml_add(request, F_CIB_CLIENTID, private->id);

	if (pcmk_is_set(call_options, cib_transaction)) {
	rc = cib_file_extend_transaction(cib, operation, request);

	if (rc != pcmk_rc_ok) {
	crm_warn("Could not extend transaction for CIB file client: %s",
	pcmk_rc_str(rc));
	}
	rc = pcmk_rc2legacy(rc);
	goto done;
	}

	rc = cib_file_process_request(cib, request, &output);

	if ((output_data != NULL) && (output != NULL)) {
	if (output->doc == private->cib_xml->doc) {
	*output_data = copy_xml(output);
	} else {
	*output_data = output;
	}
	}

	done:
	if ((output != NULL)
	&& (output->doc != private->cib_xml->doc)
	&& ((output_data == NULL) \|\| (output != *output_data))) {

	free_xml(output);
	}
	free_xml(request);
	return rc;
	}

	/*!
	* \internal
	* \brief Read CIB from disk and validate it against XML schema
	*
	* \param[in] filename Name of file to read CIB from
	* \param[out] output Where to store the read CIB XML
	*
	* \return pcmk_ok on success,
	* -ENXIO if file does not exist (or stat() otherwise fails), or
	* -pcmk_err_schema_validation if XML doesn't parse or validate
	* \note If filename is the live CIB, this will not verify its digest,
	* though that functionality would be trivial to add here.
	* Also, this will not verify that the file is writable,
	* because some callers might not need to write.
	*/
	static int
	load_file_cib(const char filename, xmlNode *output)
	{
	struct stat buf;
	xmlNode *root = NULL;

	/* Ensure file is readable */
	if (strcmp(filename, "-") && (stat(filename, &buf) < 0)) {
	return -ENXIO;
	}

	/* Parse XML from file */
	root = filename2xml(filename);
	if (root == NULL) {
	return -pcmk_err_schema_validation;
	}

	/* Add a status section if not already present */
	if (find_xml_node(root, XML_CIB_TAG_STATUS, FALSE) == NULL) {
	create_xml_node(root, XML_CIB_TAG_STATUS);
	}

	/* Validate XML against its specified schema */
	if (validate_xml(root, NULL, TRUE) == FALSE) {
	const char *schema = crm_element_value(root, XML_ATTR_VALIDATION);

	crm_err("CIB does not validate against %s", schema);
	free_xml(root);
	return -pcmk_err_schema_validation;
	}

	/* Remember the parsed XML for later use */
	*output = root;
	return pcmk_ok;
	}

	static int
	cib_file_signon(cib_t cib, const char name, enum cib_conn_type type)
	{
	int rc = pcmk_ok;
	cib_file_opaque_t *private = cib->variant_opaque;

	if (private->filename == NULL) {
	rc = -EINVAL;
	} else {
	rc = load_file_cib(private->filename, &private->cib_xml);
	}

	if (rc == pcmk_ok) {
	crm_debug("Opened connection to local file '%s' for %s",
	private->filename, name);
	cib->state = cib_connected_command;
	cib->type = cib_command;
	register_client(cib);

	} else {
	crm_info("Connection to local file '%s' for %s (client %s) failed: %s",
	private->filename, name, private->id, pcmk_strerror(rc));
	}
	return rc;
	}

	/*!
	* \internal
	* \brief Write out the in-memory CIB to a live CIB file
	*
	* param[in] cib_root Root of XML tree to write
	* param[in,out] path Full path to file to write
	*
	* \return 0 on success, -1 on failure
	*/
	static int
	cib_file_write_live(xmlNode cib_root, char path)
	{
	uid_t uid = geteuid();
	struct passwd *daemon_pwent;
	char *sep = strrchr(path, '/');
	const char cib_dirname, cib_filename;
	int rc = 0;

	/* Get the desired uid/gid */
	errno = 0;
	daemon_pwent = getpwnam(CRM_DAEMON_USER);
	if (daemon_pwent == NULL) {
	crm_perror(LOG_ERR, "Could not find %s user", CRM_DAEMON_USER);
	return -1;
	}

	/* If we're root, we can change the ownership;
	* if we're daemon, anything we create will be OK;
	* otherwise, block access so we don't create wrong owner
	*/
	if ((uid != 0) && (uid != daemon_pwent->pw_uid)) {
	crm_perror(LOG_ERR, "Must be root or %s to modify live CIB",
	CRM_DAEMON_USER);
	return 0;
	}

	/* fancy footwork to separate dirname from filename
	* (we know the canonical name maps to the live CIB,
	* but the given name might be relative, or symlinked)
	*/
	if (sep == NULL) { /* no directory component specified */
	cib_dirname = "./";
	cib_filename = path;
	} else if (sep == path) { /* given name is in / */
	cib_dirname = "/";
	cib_filename = path + 1;
	} else { /* typical case; split given name into parts */
	*sep = '\0';
	cib_dirname = path;
	cib_filename = sep + 1;
	}

	/* if we're root, we want to update the file ownership */
	if (uid == 0) {
	cib_file_owner = daemon_pwent->pw_uid;
	cib_file_group = daemon_pwent->pw_gid;
	cib_do_chown = TRUE;
	}

	/* write the file */
	if (cib_file_write_with_digest(cib_root, cib_dirname,
	cib_filename) != pcmk_ok) {
	rc = -1;
	}

	/* turn off file ownership changes, for other callers */
	if (uid == 0) {
	cib_do_chown = FALSE;
	}

	/* undo fancy stuff */
	if ((sep != NULL) && (*sep == '\0')) {
	*sep = '/';
	}

	return rc;
	}

	/*!
	* \internal
	* \brief Sign-off method for CIB file variants
	*
	* This will write the file to disk if needed, and free the in-memory CIB. If
	* the file is the live CIB, it will compute and write a signature as well.
	*
	* \param[in,out] cib CIB object to sign off
	*
	* \return pcmk_ok on success, pcmk_err_generic on failure
	* \todo This method should refuse to write the live CIB if the CIB manager is
	* running.
	*/
	static int
	cib_file_signoff(cib_t *cib)
	{
	int rc = pcmk_ok;
	cib_file_opaque_t *private = cib->variant_opaque;

	crm_debug("Disconnecting from the CIB manager");
	cib->state = cib_disconnected;
	cib->type = cib_no_connection;
	unregister_client(cib);
	cib_file_discard_transaction(cib);

	/* If the in-memory CIB has been changed, write it to disk */
	if (pcmk_is_set(private->flags, cib_file_flag_dirty)) {

	/* If this is the live CIB, write it out with a digest */
	if (pcmk_is_set(private->flags, cib_file_flag_live)) {
	if (cib_file_write_live(private->cib_xml, private->filename) < 0) {
	rc = pcmk_err_generic;
	}

	/* Otherwise, it's a simple write */
	} else {
	gboolean do_bzip = pcmk__ends_with_ext(private->filename, ".bz2");

	if (write_xml_file(private->cib_xml, private->filename,
	do_bzip) <= 0) {
	rc = pcmk_err_generic;
	}
	}

	if (rc == pcmk_ok) {
	crm_info("Wrote CIB to %s", private->filename);
	cib_clear_file_flags(private, cib_file_flag_dirty);
	} else {
	crm_err("Could not write CIB to %s", private->filename);
	}
	}

	/* Free the in-memory CIB */
	free_xml(private->cib_xml);
	private->cib_xml = NULL;
	return rc;
	}

	static int
	cib_file_free(cib_t *cib)
	{
	int rc = pcmk_ok;

	if (cib->state != cib_disconnected) {
	rc = cib_file_signoff(cib);
	}

	if (rc == pcmk_ok) {
	cib_file_opaque_t *private = cib->variant_opaque;

	free(private->id);
	free(private->filename);
	free(private);
	free(cib->cmds);
	free(cib);

	} else {
	fprintf(stderr, "Couldn't sign off: %d\n", rc);
	}

	return rc;
	}

	static int
	cib_file_inputfd(cib_t *cib)
	{
	return -EPROTONOSUPPORT;
	}

	static int
	cib_file_register_notification(cib_t cib, const char callback, int enabled)
	{
	return -EPROTONOSUPPORT;
	}

	static int
	cib_file_set_connection_dnotify(cib_t *cib,
	void (*dnotify) (gpointer user_data))
	{
	return -EPROTONOSUPPORT;
	}

	/*!
	* \internal
	* \brief Get the given CIB connection's unique client identifier
	*
	* \param[in] cib CIB connection
	* \param[out] async_id If not \p NULL, where to store asynchronous client ID
	* \param[out] sync_id If not \p NULL, where to store synchronous client ID
	*
	* \return Legacy Pacemaker return code
	*
	* \note This is the \p cib_file variant implementation of
	* \p cib_api_operations_t:client_id().
	*/
	static int
	cib_file_client_id(const cib_t cib, const char *async_id,
	const char **sync_id)
	{
	cib_file_opaque_t *private = cib->variant_opaque;

	if (async_id != NULL) {
	*async_id = private->id;
	}
	if (sync_id != NULL) {
	*sync_id = private->id;
	}
	return pcmk_ok;
	}

	cib_t *
	cib_file_new(const char *cib_location)
	{
	cib_file_opaque_t *private = NULL;
	cib_t *cib = cib_new_variant();

	if (cib == NULL) {
	return NULL;
	}

	private = calloc(1, sizeof(cib_file_opaque_t));

	if (private == NULL) {
	free(cib);
	return NULL;
	}
	private->id = crm_generate_uuid();

	cib->variant = cib_file;
	cib->variant_opaque = private;

	if (cib_location == NULL) {
	cib_location = getenv("CIB_file");
	CRM_CHECK(cib_location != NULL, return NULL); // Shouldn't be possible
	}
	private->flags = 0;
	if (cib_file_is_live(cib_location)) {
	cib_set_file_flags(private, cib_file_flag_live);
	crm_trace("File %s detected as live CIB", cib_location);
	}
	private->filename = strdup(cib_location);

	/* assign variant specific ops */
	cib->delegate_fn = cib_file_perform_op_delegate;
	cib->cmds->signon = cib_file_signon;
	cib->cmds->signoff = cib_file_signoff;
	cib->cmds->free = cib_file_free;
	cib->cmds->inputfd = cib_file_inputfd; // Deprecated method

	cib->cmds->register_notification = cib_file_register_notification;
	cib->cmds->set_connection_dnotify = cib_file_set_connection_dnotify;

	cib->cmds->client_id = cib_file_client_id;

	return cib;
	}

	/*!
	* \internal
	* \brief Compare the calculated digest of an XML tree against a signature file
	*
	* \param[in] root Root of XML tree to compare
	* \param[in] sigfile Name of signature file containing digest to compare
	*
	* \return TRUE if digests match or signature file does not exist, else FALSE
	*/
	static gboolean
	cib_file_verify_digest(xmlNode root, const char sigfile)
	{
	gboolean passed = FALSE;
	char *expected;
	int rc = pcmk__file_contents(sigfile, &expected);

	switch (rc) {
	case pcmk_rc_ok:
	if (expected == NULL) {
	crm_err("On-disk digest at %s is empty", sigfile);
	return FALSE;
	}
	break;
	case ENOENT:
	crm_warn("No on-disk digest present at %s", sigfile);
	return TRUE;
	default:
	crm_err("Could not read on-disk digest from %s: %s",
	sigfile, pcmk_rc_str(rc));
	return FALSE;
	}
	passed = pcmk__verify_digest(root, expected);
	free(expected);
	return passed;
	}

	/*!
	* \internal
	* \brief Read an XML tree from a file and verify its digest
	*
	* \param[in] filename Name of XML file to read
	* \param[in] sigfile Name of signature file containing digest to compare
	* \param[out] root If non-NULL, will be set to pointer to parsed XML tree
	*
	* \return 0 if file was successfully read, parsed and verified, otherwise:
	* -errno on stat() failure,
	* -pcmk_err_cib_corrupt if file size is 0 or XML is not parseable, or
	* -pcmk_err_cib_modified if digests do not match
	* \note If root is non-NULL, it is the caller's responsibility to free *root on
	* successful return.
	*/
	int
	cib_file_read_and_verify(const char filename, const char sigfile, xmlNode **root)
	{
	int s_res;
	struct stat buf;
	char *local_sigfile = NULL;
	xmlNode *local_root = NULL;

	CRM_ASSERT(filename != NULL);
	if (root) {
	*root = NULL;
	}

	/* Verify that file exists and its size is nonzero */
	s_res = stat(filename, &buf);
	if (s_res < 0) {
	crm_perror(LOG_WARNING, "Could not verify cluster configuration file %s", filename);
	return -errno;
	} else if (buf.st_size == 0) {
	crm_warn("Cluster configuration file %s is corrupt (size is zero)", filename);
	return -pcmk_err_cib_corrupt;
	}

	/* Parse XML */
	local_root = filename2xml(filename);
	if (local_root == NULL) {
	crm_warn("Cluster configuration file %s is corrupt (unparseable as XML)", filename);
	return -pcmk_err_cib_corrupt;
	}

	/* If sigfile is not specified, use original file name plus .sig */
	if (sigfile == NULL) {
	sigfile = local_sigfile = crm_strdup_printf("%s.sig", filename);
	}

	/* Verify that digests match */
	if (cib_file_verify_digest(local_root, sigfile) == FALSE) {
	free(local_sigfile);
	free_xml(local_root);
	return -pcmk_err_cib_modified;
	}

	free(local_sigfile);
	if (root) {
	*root = local_root;
	} else {
	free_xml(local_root);
	}
	return pcmk_ok;
	}

	/*!
	* \internal
	* \brief Back up a CIB
	*
	* \param[in] cib_dirname Directory containing CIB file and backups
	* \param[in] cib_filename Name (relative to cib_dirname) of CIB file to back up
	*
	* \return 0 on success, -1 on error
	*/
	static int
	cib_file_backup(const char cib_dirname, const char cib_filename)
	{
	int rc = 0;
	unsigned int seq;
	char *cib_path = crm_strdup_printf("%s/%s", cib_dirname, cib_filename);
	char *cib_digest = crm_strdup_printf("%s.sig", cib_path);
	char *backup_path;
	char *backup_digest;

	// Determine backup and digest file names
	if (pcmk__read_series_sequence(cib_dirname, CIB_SERIES,
	&seq) != pcmk_rc_ok) {
	// @TODO maybe handle errors better ...
	seq = 0;
	}
	backup_path = pcmk__series_filename(cib_dirname, CIB_SERIES, seq,
	CIB_SERIES_BZIP);
	backup_digest = crm_strdup_printf("%s.sig", backup_path);

	/* Remove the old backups if they exist */
	unlink(backup_path);
	unlink(backup_digest);

	/* Back up the CIB, by hard-linking it to the backup name */
	if ((link(cib_path, backup_path) < 0) && (errno != ENOENT)) {
	crm_perror(LOG_ERR, "Could not archive %s by linking to %s",
	cib_path, backup_path);
	rc = -1;

	/* Back up the CIB signature similarly */
	} else if ((link(cib_digest, backup_digest) < 0) && (errno != ENOENT)) {
	crm_perror(LOG_ERR, "Could not archive %s by linking to %s",
	cib_digest, backup_digest);
	rc = -1;

	/* Update the last counter and ensure everything is sync'd to media */
	} else {
	pcmk__write_series_sequence(cib_dirname, CIB_SERIES, ++seq,
	CIB_SERIES_MAX);
	if (cib_do_chown) {
	int rc2;

	if ((chown(backup_path, cib_file_owner, cib_file_group) < 0)
	&& (errno != ENOENT)) {
	crm_perror(LOG_ERR, "Could not set owner of %s", backup_path);
	rc = -1;
	}
	if ((chown(backup_digest, cib_file_owner, cib_file_group) < 0)
	&& (errno != ENOENT)) {
	crm_perror(LOG_ERR, "Could not set owner of %s", backup_digest);
	rc = -1;
	}
	rc2 = pcmk__chown_series_sequence(cib_dirname, CIB_SERIES,
	cib_file_owner, cib_file_group);
	if (rc2 != pcmk_rc_ok) {
	crm_err("Could not set owner of sequence file in %s: %s",
	cib_dirname, pcmk_rc_str(rc2));
	rc = -1;
	}
	}
	pcmk__sync_directory(cib_dirname);
	crm_info("Archived previous version as %s", backup_path);
	}

	free(cib_path);
	free(cib_digest);
	free(backup_path);
	free(backup_digest);
	return rc;
	}

	/*!
	* \internal
	* \brief Prepare CIB XML to be written to disk
	*
	* Set num_updates to 0, set cib-last-written to the current timestamp,
	* and strip out the status section.
	*
	* \param[in,out] root Root of CIB XML tree
	*
	* \return void
	*/
	static void
	cib_file_prepare_xml(xmlNode *root)
	{
	xmlNode *cib_status_root = NULL;

	/* Always write out with num_updates=0 and current last-written timestamp */
	crm_xml_add(root, XML_ATTR_NUMUPDATES, "0");
	pcmk__xe_add_last_written(root);

	/* Delete status section before writing to file, because
	* we discard it on startup anyway, and users get confused by it */
	cib_status_root = find_xml_node(root, XML_CIB_TAG_STATUS, TRUE);
	CRM_LOG_ASSERT(cib_status_root != NULL);
	if (cib_status_root != NULL) {
	free_xml(cib_status_root);
	}
	}

	/*!
	* \internal
	* \brief Write CIB to disk, along with a signature file containing its digest
	*
	* \param[in,out] cib_root Root of XML tree to write
	* \param[in] cib_dirname Directory containing CIB and signature files
	* \param[in] cib_filename Name (relative to cib_dirname) of file to write
	*
	* \return pcmk_ok on success,
	* pcmk_err_cib_modified if existing cib_filename doesn't match digest,
	* pcmk_err_cib_backup if existing cib_filename couldn't be backed up,
	* or pcmk_err_cib_save if new cib_filename couldn't be saved
	*/
	int
	cib_file_write_with_digest(xmlNode cib_root, const char cib_dirname,
	const char *cib_filename)
	{
	int exit_rc = pcmk_ok;
	int rc, fd;
	char *digest = NULL;

	/* Detect CIB version for diagnostic purposes */
	const char *epoch = crm_element_value(cib_root, XML_ATTR_GENERATION);
	const char *admin_epoch = crm_element_value(cib_root,
	XML_ATTR_GENERATION_ADMIN);

	/* Determine full CIB and signature pathnames */
	char *cib_path = crm_strdup_printf("%s/%s", cib_dirname, cib_filename);
	char *digest_path = crm_strdup_printf("%s.sig", cib_path);

	/* Create temporary file name patterns for writing out CIB and signature */
	char *tmp_cib = crm_strdup_printf("%s/cib.XXXXXX", cib_dirname);
	char *tmp_digest = crm_strdup_printf("%s/cib.XXXXXX", cib_dirname);

	CRM_ASSERT((cib_path != NULL) && (digest_path != NULL)
	&& (tmp_cib != NULL) && (tmp_digest != NULL));

	/* Ensure the admin didn't modify the existing CIB underneath us */
	crm_trace("Reading cluster configuration file %s", cib_path);
	rc = cib_file_read_and_verify(cib_path, NULL, NULL);
	if ((rc != pcmk_ok) && (rc != -ENOENT)) {
	crm_err("%s was manually modified while the cluster was active!",
	cib_path);
	exit_rc = pcmk_err_cib_modified;
	goto cleanup;
	}

	/* Back up the existing CIB */
	if (cib_file_backup(cib_dirname, cib_filename) < 0) {
	exit_rc = pcmk_err_cib_backup;
	goto cleanup;
	}

	crm_debug("Writing CIB to disk");
	umask(S_IWGRP \| S_IWOTH \| S_IROTH);
	cib_file_prepare_xml(cib_root);

	/* Write the CIB to a temporary file, so we can deploy (near) atomically */
	fd = mkstemp(tmp_cib);
	if (fd < 0) {
	crm_perror(LOG_ERR, "Couldn't open temporary file %s for writing CIB",
	tmp_cib);
	exit_rc = pcmk_err_cib_save;
	goto cleanup;
	}

	/* Protect the temporary file */
	if (fchmod(fd, S_IRUSR \| S_IWUSR) < 0) {
	crm_perror(LOG_ERR, "Couldn't protect temporary file %s for writing CIB",
	tmp_cib);
	exit_rc = pcmk_err_cib_save;
	goto cleanup;
	}
	if (cib_do_chown && (fchown(fd, cib_file_owner, cib_file_group) < 0)) {
	crm_perror(LOG_ERR, "Couldn't protect temporary file %s for writing CIB",
	tmp_cib);
	exit_rc = pcmk_err_cib_save;
	goto cleanup;
	}

	/* Write out the CIB */
	if (write_xml_fd(cib_root, tmp_cib, fd, FALSE) <= 0) {
	crm_err("Changes couldn't be written to %s", tmp_cib);
	exit_rc = pcmk_err_cib_save;
	goto cleanup;
	}

	/* Calculate CIB digest */
	digest = calculate_on_disk_digest(cib_root);
	CRM_ASSERT(digest != NULL);
	crm_info("Wrote version %s.%s.0 of the CIB to disk (digest: %s)",
	(admin_epoch ? admin_epoch : "0"), (epoch ? epoch : "0"), digest);

	/* Write the CIB digest to a temporary file */
	fd = mkstemp(tmp_digest);
	if (fd < 0) {
	crm_perror(LOG_ERR, "Could not create temporary file for CIB digest");
	exit_rc = pcmk_err_cib_save;
	goto cleanup;
	}
	if (cib_do_chown && (fchown(fd, cib_file_owner, cib_file_group) < 0)) {
	crm_perror(LOG_ERR, "Couldn't protect temporary file %s for writing CIB",
	tmp_cib);
	exit_rc = pcmk_err_cib_save;
	close(fd);
	goto cleanup;
	}
	rc = pcmk__write_sync(fd, digest);
	if (rc != pcmk_rc_ok) {
	crm_err("Could not write digest to %s: %s",
	tmp_digest, pcmk_rc_str(rc));
	exit_rc = pcmk_err_cib_save;
	close(fd);
	goto cleanup;
	}
	close(fd);
	crm_debug("Wrote digest %s to disk", digest);

	/* Verify that what we wrote is sane */
	crm_info("Reading cluster configuration file %s (digest: %s)",
	tmp_cib, tmp_digest);
	rc = cib_file_read_and_verify(tmp_cib, tmp_digest, NULL);
	CRM_ASSERT(rc == 0);

	/* Rename temporary files to live, and sync directory changes to media */
	crm_debug("Activating %s", tmp_cib);
	if (rename(tmp_cib, cib_path) < 0) {
	crm_perror(LOG_ERR, "Couldn't rename %s as %s", tmp_cib, cib_path);
	exit_rc = pcmk_err_cib_save;
	}
	if (rename(tmp_digest, digest_path) < 0) {
	crm_perror(LOG_ERR, "Couldn't rename %s as %s", tmp_digest,
	digest_path);
	exit_rc = pcmk_err_cib_save;
	}
	pcmk__sync_directory(cib_dirname);

	cleanup:
	free(cib_path);
	free(digest_path);
	free(digest);
	free(tmp_digest);
	free(tmp_cib);
	return exit_rc;
	}

	/*!
	* \internal
	* \brief Create a new transaction for a given CIB file client
	*
	* \param[in,out] cib CIB file client
	*
	* \return Standard Pacemaker return code
	*/
	static int
	cib_file_init_transaction(cib_t *cib)
	{
	cib_file_opaque_t *private = cib->variant_opaque;

	// A client can have at most one transaction at a time
	if (private->transaction != NULL) {
	return pcmk_rc_already;
	}

	crm_trace("Initiating transaction for CIB file client (%s) on file '%s'",
	private->id, private->filename);

	private->transaction = g_queue_new();
	return pcmk_rc_ok;
	}

	/*!
	* \internal
	* \brief Add a new CIB request to an existing transaction
	*
	* \param[in,out] cib CIB file client
	* \param[in] operation CIB operation
	* \param[in] request CIB request
	*
	* \return Standard Pacemaker return code
	*/
	static int
	cib_file_extend_transaction(cib_t cib, const cib_operation_t operation,
	xmlNode *request)
	{
	cib_file_opaque_t *private = cib->variant_opaque;

	if (private->transaction == NULL) {
	return pcmk_rc_no_transaction;
	}
	-
	if (!pcmk_is_set(operation->flags, cib_op_attr_transaction)) {
	crm_err("Operation '%s' is not supported in CIB transaction",
	operation->name);
	return EOPNOTSUPP;
	}
	-
	+ if (file_get_op_function(operation) == NULL) {
	+ crm_err("Operation %s is not supported by CIB file clients",
	+ operation->name);
	+ return EOPNOTSUPP;
	+ }
	g_queue_push_tail(private->transaction, copy_xml(request));
	return pcmk_rc_ok;
	}

	/*!
	* \internal
	* \brief Free a CIB file client's transaction (if any) and its requests
	*
	* \param[in,out] cib CIB file client
	*/
	static void
	cib_file_discard_transaction(cib_t *cib)
	{
	bool found = false;
	cib_file_opaque_t *private = cib->variant_opaque;

	if (private->transaction != NULL) {
	found = true;
	g_queue_free_full(private->transaction, (GDestroyNotify) free_xml);
	private->transaction = NULL;
	}

	crm_trace("%s for CIB file client (%s) on file '%s'",
	(found? "Discarded transaction" : "No transaction found"),
	private->id, private->filename);
	}

	/*!
	* \internal
	* \brief Process requests in a CIB file client's transaction
	*
	* Stop when a request fails or when all requests have been processed.
	*
	* \param[in,out] cib CIB file client
	*
	* \return Standard Pacemaker return code
	*/
	static int
	cib_file_process_transaction_requests(cib_t *cib)
	{
	cib_file_opaque_t *private = cib->variant_opaque;
	GQueue *transaction = private->transaction;

	for (xmlNode *request = g_queue_pop_head(transaction); request != NULL;
	request = g_queue_pop_head(transaction)) {

	xmlNode *output = NULL;
	const char *op = crm_element_value(request, F_CIB_OPERATION);

	int rc = cib_file_process_request(cib, request, &output);

	rc = pcmk_legacy2rc(rc);
	if (rc != pcmk_rc_ok) {
	crm_err("Aborting transaction for CIB file client (%s) on file "
	"'%s' due to failed %s request: %s",
	private->id, private->filename, op, pcmk_rc_str(rc));
	crm_log_xml_info(request, "Failed request");
	free_xml(request);
	return rc;
	}

	crm_trace("Applied %s request to transaction working CIB for CIB file "
	"client (%s) on file '%s'",
	op, private->id, private->filename);
	crm_log_xml_trace(request, "Successful request");
	free_xml(request);
	}

	return pcmk_rc_ok;
	}

	/*!
	* \internal
	* \brief Commit a given CIB file client's transaction to a working CIB copy
	*
	* \param[in,out] cib CIB file client
	* \param[in,out] result_cib Where to store result CIB
	*
	* \return Standard Pacemaker return code
	*
	* \note The caller is responsible for replacing the \p cib argument's
	* \p private->cib_xml with \p result_cib on success, and for freeing
	* \p result_cib using \p free_xml() on failure.
	*/
	static int
	cib_file_commit_transaction(cib_t cib, xmlNode *result_cib)
	{
	int rc = pcmk_rc_ok;
	cib_file_opaque_t *private = cib->variant_opaque;
	xmlNode *saved_cib = private->cib_xml;

	/* *result_cib should be a copy of private->cib_xml (created by
	* cib_perform_op()). If not, make a copy now. Change tracking isn't
	* strictly required here because:
	* * Each request in the transaction will have changes tracked and ACLs
	* checked if appropriate.
	* * cib_perform_op() will infer changes for the commit request at the end.
	*/
	CRM_CHECK((result_cib != NULL) && (result_cib != private->cib_xml),
	*result_cib = copy_xml(private->cib_xml));

	if (private->transaction == NULL) {
	return pcmk_rc_no_transaction;
	}

	crm_trace("Committing transaction for CIB file client (%s) on file '%s' to "
	"working CIB",
	private->id, private->filename);

	// Apply all changes to a working copy of the CIB
	private->cib_xml = *result_cib;

	rc = cib_file_process_transaction_requests(cib);

	crm_trace("Transaction commit %s for CIB file client (%s) on file '%s'; "
	"discarding queue",
	((rc != pcmk_rc_ok)? "succeeded" : "failed"),
	private->id, private->filename);

	// Free the transaction and (if aborted) free any remaining requests
	cib_file_discard_transaction(cib);

	/* Some request types (for example, erase) may have freed private->cib_xml
	* (the working copy) and pointed it at a new XML object. In that case, it
	* follows that *result_cib (the working copy) was freed.
	*
	* Point *result_cib at the updated working copy stored in private->cib_xml.
	*/
	*result_cib = private->cib_xml;

	// Point private->cib_xml back to the unchanged original copy
	private->cib_xml = saved_cib;

	return rc;
	}

	static int
	cib_file_process_init_transaction(const char *op, int options,
	const char section, xmlNode req,
	xmlNode input, xmlNode existing_cib,
	xmlNode result_cib, xmlNode answer)
	{
	int rc = pcmk_rc_ok;
	const char *client_id = crm_element_value(req, F_CIB_CLIENTID);
	cib_t *cib = NULL;

	CRM_CHECK(client_id != NULL, return -EINVAL);

	cib = get_client(client_id);
	CRM_CHECK(cib != NULL, return -EINVAL);

	rc = cib_file_init_transaction(cib);

	if (rc != pcmk_rc_ok) {
	cib_file_opaque_t *private = cib->variant_opaque;

	crm_err("Could not initiate transaction for CIB file client (%s) on "
	"file '%s': %s",
	private->id, private->filename, pcmk_rc_str(rc));
	}
	return pcmk_rc2legacy(rc);
	}

	static int
	cib_file_process_commit_transaction(const char *op, int options,
	const char section, xmlNode req,
	xmlNode input, xmlNode existing_cib,
	xmlNode result_cib, xmlNode answer)
	{
	int rc = pcmk_rc_ok;
	const char *client_id = crm_element_value(req, F_CIB_CLIENTID);
	cib_t *cib = NULL;

	CRM_CHECK(client_id != NULL, return -EINVAL);

	cib = get_client(client_id);
	CRM_CHECK(cib != NULL, return -EINVAL);

	rc = cib_file_commit_transaction(cib, result_cib);
	if (rc != pcmk_rc_ok) {
	cib_file_opaque_t *private = cib->variant_opaque;

	crm_err("Could not commit transaction for CIB file client (%s) on "
	"file '%s': %s",
	private->id, private->filename, pcmk_rc_str(rc));
	}
	return pcmk_rc2legacy(rc);
	}

	static int
	cib_file_process_discard_transaction(const char *op, int options,
	const char section, xmlNode req,
	xmlNode input, xmlNode existing_cib,
	xmlNode result_cib, xmlNode answer)
	{
	const char *client_id = crm_element_value(req, F_CIB_CLIENTID);
	cib_t *cib = NULL;

	CRM_CHECK(client_id != NULL, return -EINVAL);

	cib = get_client(client_id);
	CRM_CHECK(cib != NULL, return -EINVAL);

	cib_file_discard_transaction(cib);
	return pcmk_ok;
	}
	diff --git a/lib/cib/cib_ops.c b/lib/cib/cib_ops.c
	index 3be0649b97..9cd962deb6 100644
	--- a/lib/cib/cib_ops.c
	+++ b/lib/cib/cib_ops.c
	@@ -1,876 +1,1025 @@
	/*
	* Copyright 2004-2023 the Pacemaker project contributors
	*
	* The version control history for this file may have further details.
	*
	* This source code is licensed under the GNU Lesser General Public License
	* version 2.1 or later (LGPLv2.1+) WITHOUT ANY WARRANTY.
	*/

	#include <crm_internal.h>

	#include <stdio.h>
	#include <unistd.h>
	#include <stdlib.h>
	#include <errno.h>
	#include <fcntl.h>
	#include <time.h>

	#include <sys/param.h>
	#include <sys/types.h>

	+#include <glib.h>
	+#include <libxml/tree.h>
	+
	#include <crm/crm.h>
	#include <crm/cib/internal.h>
	#include <crm/msg_xml.h>

	#include <crm/common/xml.h>
	#include <crm/common/xml_internal.h>

	+// @TODO: Free this via crm_exit() when libcib gets merged with libcrmcommon
	+static GHashTable *operation_table = NULL;
	+
	+static const cib_operation_t cib_ops[] = {
	+ {
	+ PCMK__CIB_REQUEST_ABS_DELETE, cib__op_abs_delete,
	+ cib_op_attr_modifies\|cib_op_attr_privileged
	+ },
	+ {
	+ PCMK__CIB_REQUEST_APPLY_PATCH, cib__op_apply_patch,
	+ cib_op_attr_modifies
	+ \|cib_op_attr_privileged
	+ \|cib_op_attr_transaction
	+ },
	+ {
	+ PCMK__CIB_REQUEST_BUMP, cib__op_bump,
	+ cib_op_attr_modifies
	+ \|cib_op_attr_privileged
	+ \|cib_op_attr_transaction
	+ },
	+ {
	+ PCMK__CIB_REQUEST_CREATE, cib__op_create,
	+ cib_op_attr_modifies
	+ \|cib_op_attr_privileged
	+ \|cib_op_attr_transaction
	+ },
	+ {
	+ PCMK__CIB_REQUEST_DELETE, cib__op_delete,
	+ cib_op_attr_modifies
	+ \|cib_op_attr_privileged
	+ \|cib_op_attr_transaction
	+ },
	+ {
	+ PCMK__CIB_REQUEST_ERASE, cib__op_erase,
	+ cib_op_attr_modifies
	+ \|cib_op_attr_privileged
	+ \|cib_op_attr_replaces
	+ \|cib_op_attr_transaction
	+ },
	+ {
	+ PCMK__CIB_REQUEST_IS_PRIMARY, cib__op_is_primary, cib_op_attr_privileged
	+ },
	+ {
	+ PCMK__CIB_REQUEST_MODIFY, cib__op_modify,
	+ cib_op_attr_modifies
	+ \|cib_op_attr_privileged
	+ \|cib_op_attr_transaction
	+ },
	+ {
	+ PCMK__CIB_REQUEST_NOOP, cib__op_noop, cib_op_attr_none
	+ },
	+ {
	+ CRM_OP_PING, cib__op_ping, cib_op_attr_none
	+ },
	+ {
	+ // @COMPAT: Drop cib_op_attr_modifies when we drop legacy mode support
	+ PCMK__CIB_REQUEST_PRIMARY, cib__op_primary,
	+ cib_op_attr_modifies\|cib_op_attr_privileged\|cib_op_attr_local
	+ },
	+ {
	+ PCMK__CIB_REQUEST_QUERY, cib__op_query, cib_op_attr_none
	+ },
	+ {
	+ PCMK__CIB_REQUEST_REPLACE, cib__op_replace,
	+ cib_op_attr_modifies
	+ \|cib_op_attr_privileged
	+ \|cib_op_attr_replaces
	+ \|cib_op_attr_writes_through
	+ \|cib_op_attr_transaction
	+ },
	+ {
	+ PCMK__CIB_REQUEST_SECONDARY, cib__op_secondary,
	+ cib_op_attr_privileged\|cib_op_attr_local
	+ },
	+ {
	+ PCMK__CIB_REQUEST_SHUTDOWN, cib__op_shutdown, cib_op_attr_privileged
	+ },
	+ {
	+ PCMK__CIB_REQUEST_SYNC_TO_ALL, cib__op_sync_all, cib_op_attr_privileged
	+ },
	+ {
	+ PCMK__CIB_REQUEST_SYNC_TO_ONE, cib__op_sync_one, cib_op_attr_privileged
	+ },
	+ {
	+ PCMK__CIB_REQUEST_UPGRADE, cib__op_upgrade,
	+ cib_op_attr_modifies
	+ \|cib_op_attr_privileged
	+ \|cib_op_attr_writes_through
	+ \|cib_op_attr_transaction
	+ },
	+
	+ /* PCMK__CIB_REQUEST_*_TRANSACT requests must be processed locally because
	+ * they depend on the client table. Requests that manage transactions on
	+ * other nodes would likely be problematic in many other ways as well.
	+ */
	+ {
	+ PCMK__CIB_REQUEST_INIT_TRANSACT, cib__op_init_transact,
	+ cib_op_attr_privileged\|cib_op_attr_local
	+ },
	+ {
	+ PCMK__CIB_REQUEST_COMMIT_TRANSACT, cib__op_commit_transact,
	+ cib_op_attr_modifies
	+ \|cib_op_attr_privileged
	+ \|cib_op_attr_local
	+ \|cib_op_attr_replaces
	+ \|cib_op_attr_writes_through
	+ },
	+ {
	+ PCMK__CIB_REQUEST_DISCARD_TRANSACT, cib__op_discard_transact,
	+ cib_op_attr_privileged\|cib_op_attr_local
	+ },
	+};
	+
	+/*!
	+ * \internal
	+ * \brief Get the \c cib_operation_t object for a given CIB operation name
	+ *
	+ * \param[in] op CIB operation name
	+ * \param[out] operation Where to store CIB operation object
	+ *
	+ * \return Standard Pacemaker return code
	+ */
	+int
	+cib__get_operation(const char op, const cib_operation_t *operation)
	+{
	+ CRM_ASSERT((op != NULL) && (operation != NULL));
	+
	+ if (operation_table == NULL) {
	+ operation_table = pcmk__strkey_table(NULL, NULL);
	+
	+ for (int lpc = 0; lpc < PCMK__NELEM(cib_ops); lpc++) {
	+ const cib_operation_t *oper = &(cib_ops[lpc]);
	+
	+ g_hash_table_insert(operation_table, (gpointer) oper->name,
	+ (gpointer) oper);
	+ }
	+ }
	+
	+ *operation = g_hash_table_lookup(operation_table, op);
	+ if (*operation == NULL) {
	+ crm_err("Operation %s is invalid", op);
	+ return EINVAL;
	+ }
	+ return pcmk_rc_ok;
	+}
	+
	int
	cib_process_query(const char op, int options, const char section, xmlNode * req, xmlNode * input,
	xmlNode * existing_cib, xmlNode result_cib, xmlNode answer)
	{
	xmlNode *obj_root = NULL;
	int result = pcmk_ok;

	crm_trace("Processing %s for %s section",
	op, pcmk__s(section, "unspecified"));

	if (options & cib_xpath) {
	return cib_process_xpath(op, options, section, req, input,
	existing_cib, result_cib, answer);
	}

	CRM_CHECK(answer == NULL, free_xml(answer));
	*answer = NULL;

	if (pcmk__str_eq(XML_CIB_TAG_SECTION_ALL, section, pcmk__str_casei)) {
	section = NULL;
	}

	obj_root = pcmk_find_cib_element(existing_cib, section);

	if (obj_root == NULL) {
	result = -ENXIO;

	} else if (options & cib_no_children) {
	const char *tag = TYPE(obj_root);
	xmlNode shallow = create_xml_node(answer, tag);

	copy_in_properties(shallow, obj_root);
	*answer = shallow;

	} else {
	*answer = obj_root;
	}

	if (result == pcmk_ok && *answer == NULL) {
	crm_err("Error creating query response");
	result = -ENOMSG;
	}

	return result;
	}

	static int
	update_counter(xmlNode xml_obj, const char field, bool reset)
	{
	char *new_value = NULL;
	char *old_value = NULL;
	int int_value = -1;

	if (!reset && crm_element_value(xml_obj, field) != NULL) {
	old_value = crm_element_value_copy(xml_obj, field);
	}
	if (old_value != NULL) {
	int_value = atoi(old_value);
	new_value = pcmk__itoa(++int_value);
	} else {
	new_value = strdup("1");
	CRM_ASSERT(new_value != NULL);
	}

	crm_trace("Update %s from %s to %s",
	field, pcmk__s(old_value, "unset"), new_value);
	crm_xml_add(xml_obj, field, new_value);

	free(new_value);
	free(old_value);

	return pcmk_ok;
	}

	int
	cib_process_erase(const char op, int options, const char section, xmlNode * req, xmlNode * input,
	xmlNode * existing_cib, xmlNode result_cib, xmlNode answer)
	{
	int result = pcmk_ok;

	crm_trace("Processing \"%s\" event", op);

	if (*result_cib != existing_cib) {
	free_xml(*result_cib);
	}
	*result_cib = createEmptyCib(0);
	copy_in_properties(*result_cib, existing_cib);
	update_counter(*result_cib, XML_ATTR_GENERATION_ADMIN, false);
	*answer = NULL;

	return result;
	}

	int
	cib_process_upgrade(const char op, int options, const char section, xmlNode * req,
	xmlNode * input, xmlNode * existing_cib, xmlNode ** result_cib,
	xmlNode ** answer)
	{
	int rc = 0;
	int new_version = 0;
	int current_version = 0;
	int max_version = 0;
	const char *max = crm_element_value(req, F_CIB_SCHEMA_MAX);
	const char *value = crm_element_value(existing_cib, XML_ATTR_VALIDATION);

	*answer = NULL;
	crm_trace("Processing \"%s\" event with max=%s", op, max);

	if (value != NULL) {
	current_version = get_schema_version(value);
	}

	if (max) {
	max_version = get_schema_version(max);
	}

	rc = update_validation(result_cib, &new_version, max_version, TRUE,
	!(options & cib_verbose));
	if (new_version > current_version) {
	update_counter(*result_cib, XML_ATTR_GENERATION_ADMIN, false);
	update_counter(*result_cib, XML_ATTR_GENERATION, true);
	update_counter(*result_cib, XML_ATTR_NUMUPDATES, true);
	return pcmk_ok;
	}

	return rc;
	}

	int
	cib_process_bump(const char op, int options, const char section, xmlNode * req, xmlNode * input,
	xmlNode * existing_cib, xmlNode result_cib, xmlNode answer)
	{
	int result = pcmk_ok;

	crm_trace("Processing %s for epoch='%s'", op,
	pcmk__s(crm_element_value(existing_cib, XML_ATTR_GENERATION), ""));

	*answer = NULL;
	update_counter(*result_cib, XML_ATTR_GENERATION, false);

	return result;
	}

	int
	cib_process_replace(const char op, int options, const char section, xmlNode * req,
	xmlNode * input, xmlNode * existing_cib, xmlNode ** result_cib,
	xmlNode ** answer)
	{
	const char *tag = NULL;
	int result = pcmk_ok;

	crm_trace("Processing %s for %s section",
	op, pcmk__s(section, "unspecified"));

	if (options & cib_xpath) {
	return cib_process_xpath(op, options, section, req, input,
	existing_cib, result_cib, answer);
	}

	*answer = NULL;

	if (input == NULL) {
	return -EINVAL;
	}

	tag = crm_element_name(input);

	if (pcmk__str_eq(XML_CIB_TAG_SECTION_ALL, section, pcmk__str_casei)) {
	section = NULL;

	} else if (pcmk__str_eq(tag, section, pcmk__str_casei)) {
	section = NULL;
	}

	if (pcmk__str_eq(tag, XML_TAG_CIB, pcmk__str_casei)) {
	int updates = 0;
	int epoch = 0;
	int admin_epoch = 0;

	int replace_updates = 0;
	int replace_epoch = 0;
	int replace_admin_epoch = 0;

	const char *reason = NULL;
	const char *peer = crm_element_value(req, F_ORIG);
	const char *digest = crm_element_value(req, XML_ATTR_DIGEST);

	if (digest) {
	const char *version = crm_element_value(req, XML_ATTR_CRM_VERSION);
	char *digest_verify = calculate_xml_versioned_digest(input, FALSE, TRUE,
	version ? version :
	CRM_FEATURE_SET);

	if (!pcmk__str_eq(digest_verify, digest, pcmk__str_casei)) {
	crm_err("Digest mis-match on replace from %s: %s vs. %s (expected)", peer,
	digest_verify, digest);
	reason = "digest mismatch";

	} else {
	crm_info("Digest matched on replace from %s: %s", peer, digest);
	}
	free(digest_verify);

	} else {
	crm_trace("No digest to verify");
	}

	cib_version_details(existing_cib, &admin_epoch, &epoch, &updates);
	cib_version_details(input, &replace_admin_epoch, &replace_epoch, &replace_updates);

	if (replace_admin_epoch < admin_epoch) {
	reason = XML_ATTR_GENERATION_ADMIN;

	} else if (replace_admin_epoch > admin_epoch) {
	/* no more checks */

	} else if (replace_epoch < epoch) {
	reason = XML_ATTR_GENERATION;

	} else if (replace_epoch > epoch) {
	/* no more checks */

	} else if (replace_updates < updates) {
	reason = XML_ATTR_NUMUPDATES;
	}

	if (reason != NULL) {
	crm_info("Replacement %d.%d.%d from %s not applied to %d.%d.%d:"
	" current %s is greater than the replacement",
	replace_admin_epoch, replace_epoch,
	replace_updates, peer, admin_epoch, epoch, updates, reason);
	result = -pcmk_err_old_data;
	} else {
	crm_info("Replaced %d.%d.%d with %d.%d.%d from %s",
	admin_epoch, epoch, updates,
	replace_admin_epoch, replace_epoch, replace_updates, peer);
	}

	if (*result_cib != existing_cib) {
	free_xml(*result_cib);
	}
	*result_cib = copy_xml(input);

	} else {
	xmlNode *obj_root = NULL;
	gboolean ok = TRUE;

	obj_root = pcmk_find_cib_element(*result_cib, section);
	ok = replace_xml_child(NULL, obj_root, input, FALSE);
	if (ok == FALSE) {
	crm_trace("No matching object to replace");
	result = -ENXIO;
	}
	}

	return result;
	}

	int
	cib_process_delete(const char op, int options, const char section, xmlNode * req, xmlNode * input,
	xmlNode * existing_cib, xmlNode result_cib, xmlNode answer)
	{
	xmlNode *obj_root = NULL;

	crm_trace("Processing \"%s\" event", op);

	if (options & cib_xpath) {
	return cib_process_xpath(op, options, section, req, input,
	existing_cib, result_cib, answer);
	}

	if (input == NULL) {
	crm_err("Cannot perform modification with no data");
	return -EINVAL;
	}

	obj_root = pcmk_find_cib_element(*result_cib, section);
	if(pcmk__str_eq(crm_element_name(input), section, pcmk__str_casei)) {
	xmlNode *child = NULL;
	for (child = pcmk__xml_first_child(input); child;
	child = pcmk__xml_next(child)) {
	if (replace_xml_child(NULL, obj_root, child, TRUE) == FALSE) {
	crm_trace("No matching object to delete: %s=%s", child->name, ID(child));
	}
	}

	} else if (replace_xml_child(NULL, obj_root, input, TRUE) == FALSE) {
	crm_trace("No matching object to delete: %s=%s", input->name, ID(input));
	}

	return pcmk_ok;
	}

	int
	cib_process_modify(const char op, int options, const char section, xmlNode * req, xmlNode * input,
	xmlNode * existing_cib, xmlNode result_cib, xmlNode answer)
	{
	xmlNode *obj_root = NULL;

	crm_trace("Processing \"%s\" event", op);

	if (options & cib_xpath) {
	return cib_process_xpath(op, options, section, req, input,
	existing_cib, result_cib, answer);
	}

	if (input == NULL) {
	crm_err("Cannot perform modification with no data");
	return -EINVAL;
	}

	obj_root = pcmk_find_cib_element(*result_cib, section);
	if (obj_root == NULL) {
	xmlNode *tmp_section = NULL;
	const char *path = pcmk_cib_parent_name_for(section);

	if (path == NULL) {
	return -EINVAL;
	}

	tmp_section = create_xml_node(NULL, section);
	cib_process_xpath(PCMK__CIB_REQUEST_CREATE, 0, path, NULL, tmp_section,
	NULL, result_cib, answer);
	free_xml(tmp_section);

	obj_root = pcmk_find_cib_element(*result_cib, section);
	}

	CRM_CHECK(obj_root != NULL, return -EINVAL);

	if (update_xml_child(obj_root, input) == FALSE) {
	if (options & cib_can_create) {
	add_node_copy(obj_root, input);
	} else {
	return -ENXIO;
	}
	}

	if(options & cib_mixed_update) {
	int max = 0, lpc;
	xmlXPathObjectPtr xpathObj = xpath_search(*result_cib, "//@__delete__");

	if (xpathObj) {
	max = numXpathResults(xpathObj);
	crm_log_xml_trace(*result_cib, "Mixed result");
	}

	for (lpc = 0; lpc < max; lpc++) {
	xmlNode *match = getXpathResult(xpathObj, lpc);
	xmlChar *match_path = xmlGetNodePath(match);

	crm_debug("Destroying %s", match_path);
	free(match_path);
	free_xml(match);
	}

	freeXpathObject(xpathObj);
	}
	return pcmk_ok;
	}

	static int
	update_cib_object(xmlNode * parent, xmlNode * update)
	{
	int result = pcmk_ok;
	xmlNode *target = NULL;
	xmlNode *a_child = NULL;
	const char *replace = NULL;
	const char *object_id = NULL;
	const char *object_name = NULL;

	CRM_CHECK(update != NULL, return -EINVAL);
	CRM_CHECK(parent != NULL, return -EINVAL);

	object_name = crm_element_name(update);
	CRM_CHECK(object_name != NULL, return -EINVAL);

	object_id = ID(update);
	crm_trace("Processing update for <%s%s%s%s>", object_name,
	((object_id == NULL)? "" : " " XML_ATTR_ID "='"),
	pcmk__s(object_id, ""),
	((object_id == NULL)? "" : "'"));

	if (object_id == NULL) {
	/* placeholder object */
	target = find_xml_node(parent, object_name, FALSE);

	} else {
	target = pcmk__xe_match(parent, object_name, XML_ATTR_ID, object_id);
	}

	if (target == NULL) {
	target = create_xml_node(parent, object_name);
	}

	crm_trace("Found node <%s%s%s%s> to update", object_name,
	((object_id == NULL)? "" : " " XML_ATTR_ID "='"),
	pcmk__s(object_id, ""),
	((object_id == NULL)? "" : "'"));

	// @COMPAT: XML_CIB_ATTR_REPLACE is unused internally. Remove at break.
	replace = crm_element_value(update, XML_CIB_ATTR_REPLACE);
	if (replace != NULL) {
	xmlNode *remove = NULL;
	int last = 0, lpc = 0, len = 0;

	len = strlen(replace);
	while (lpc <= len) {
	if (replace[lpc] == ',' \|\| replace[lpc] == 0) {
	char *replace_item = NULL;

	if (last == lpc) {
	/* nothing to do */
	last = lpc + 1;
	goto incr;
	}

	replace_item = strndup(replace + last, lpc - last);
	remove = find_xml_node(target, replace_item, FALSE);
	if (remove != NULL) {
	crm_trace("Replacing node <%s> in <%s>",
	replace_item, crm_element_name(target));
	free_xml(remove);
	remove = NULL;
	}
	free(replace_item);
	last = lpc + 1;
	}
	incr:
	lpc++;
	}
	xml_remove_prop(update, XML_CIB_ATTR_REPLACE);
	xml_remove_prop(target, XML_CIB_ATTR_REPLACE);
	}

	copy_in_properties(target, update);

	if (xml_acl_denied(target)) {
	crm_notice("Cannot update <%s " XML_ATTR_ID "=%s>",
	pcmk__s(object_name, "<null>"),
	pcmk__s(object_id, "<null>"));
	return -EACCES;
	}

	crm_trace("Processing children of <%s%s%s%s>", object_name,
	((object_id == NULL)? "" : " " XML_ATTR_ID "='"),
	pcmk__s(object_id, ""),
	((object_id == NULL)? "" : "'"));

	for (a_child = pcmk__xml_first_child(update); a_child != NULL;
	a_child = pcmk__xml_next(a_child)) {
	int tmp_result = 0;

	crm_trace("Updating child <%s%s%s%s>", crm_element_name(a_child),
	((ID(a_child) == NULL)? "" : " " XML_ATTR_ID "='"),
	pcmk__s(ID(a_child), ""), ((ID(a_child) == NULL)? "" : "'"));

	tmp_result = update_cib_object(target, a_child);

	/* only the first error is likely to be interesting */
	if (tmp_result != pcmk_ok) {
	crm_err("Error updating child <%s%s%s%s>",
	crm_element_name(a_child),
	((ID(a_child) == NULL)? "" : " " XML_ATTR_ID "='"),
	pcmk__s(ID(a_child), ""),
	((ID(a_child) == NULL)? "" : "'"));

	if (result == pcmk_ok) {
	result = tmp_result;
	}
	}
	}

	crm_trace("Finished handling update for <%s%s%s%s>", object_name,
	((object_id == NULL)? "" : " " XML_ATTR_ID "='"),
	pcmk__s(object_id, ""),
	((object_id == NULL)? "" : "'"));

	return result;
	}

	static int
	add_cib_object(xmlNode * parent, xmlNode * new_obj)
	{
	const char *object_name = NULL;
	const char *object_id = NULL;
	xmlNode *equiv_node = NULL;

	if ((parent == NULL) \|\| (new_obj == NULL)) {
	return -EINVAL;
	}

	object_name = crm_element_name(new_obj);
	if (object_name == NULL) {
	return -EINVAL;
	}

	object_id = ID(new_obj);

	crm_trace("Processing creation of <%s%s%s%s>", object_name,
	((object_id == NULL)? "" : " " XML_ATTR_ID "='"),
	pcmk__s(object_id, ""),
	((object_id == NULL)? "" : "'"));

	if (object_id == NULL) {
	equiv_node = find_xml_node(parent, object_name, FALSE);
	} else {
	equiv_node = pcmk__xe_match(parent, object_name, XML_ATTR_ID,
	object_id);
	}
	if (equiv_node != NULL) {
	return -EEXIST;
	}

	return update_cib_object(parent, new_obj);
	}

	static bool
	update_results(xmlNode failed, xmlNode target, const char *operation,
	int return_code)
	{
	xmlNode *xml_node = NULL;
	bool was_error = false;
	const char *error_msg = NULL;

	if (return_code != pcmk_ok) {
	error_msg = pcmk_strerror(return_code);

	was_error = true;
	xml_node = create_xml_node(failed, XML_FAIL_TAG_CIB);
	add_node_copy(xml_node, target);

	crm_xml_add(xml_node, XML_FAILCIB_ATTR_ID, ID(target));
	crm_xml_add(xml_node, XML_FAILCIB_ATTR_OBJTYPE, TYPE(target));
	crm_xml_add(xml_node, XML_FAILCIB_ATTR_OP, operation);
	crm_xml_add(xml_node, XML_FAILCIB_ATTR_REASON, error_msg);

	crm_warn("Action %s failed: %s (cde=%d)",
	operation, error_msg, return_code);
	}

	return was_error;
	}

	int
	cib_process_create(const char op, int options, const char section, xmlNode * req, xmlNode * input,
	xmlNode * existing_cib, xmlNode result_cib, xmlNode answer)
	{
	xmlNode *failed = NULL;
	int result = pcmk_ok;
	xmlNode *update_section = NULL;

	crm_trace("Processing %s for %s section",
	op, pcmk__s(section, "unspecified"));
	if (pcmk__str_eq(XML_CIB_TAG_SECTION_ALL, section, pcmk__str_casei)) {
	section = NULL;

	} else if (pcmk__str_eq(XML_TAG_CIB, section, pcmk__str_casei)) {
	section = NULL;

	} else if (pcmk__str_eq(crm_element_name(input), XML_TAG_CIB, pcmk__str_casei)) {
	section = NULL;
	}

	CRM_CHECK(strcmp(op, PCMK__CIB_REQUEST_CREATE) == 0, return -EINVAL);

	if (input == NULL) {
	crm_err("Cannot perform modification with no data");
	return -EINVAL;
	}

	if (section == NULL) {
	return cib_process_modify(op, options, section, req, input, existing_cib, result_cib,
	answer);
	}

	failed = create_xml_node(NULL, XML_TAG_FAILED);

	update_section = pcmk_find_cib_element(*result_cib, section);
	if (pcmk__str_eq(crm_element_name(input), section, pcmk__str_casei)) {
	xmlNode *a_child = NULL;

	for (a_child = pcmk__xml_first_child(input); a_child != NULL;
	a_child = pcmk__xml_next(a_child)) {
	result = add_cib_object(update_section, a_child);
	if (update_results(failed, a_child, op, result)) {
	break;
	}
	}

	} else {
	result = add_cib_object(update_section, input);
	update_results(failed, input, op, result);
	}

	if ((result == pcmk_ok) && xml_has_children(failed)) {
	result = -EINVAL;
	}

	if (result != pcmk_ok) {
	crm_log_xml_err(failed, "CIB Update failures");
	*answer = failed;

	} else {
	free_xml(failed);
	}

	return result;
	}

	int
	cib_process_diff(const char op, int options, const char section, xmlNode * req, xmlNode * input,
	xmlNode * existing_cib, xmlNode result_cib, xmlNode answer)
	{
	const char *originator = NULL;

	if (req != NULL) {
	originator = crm_element_value(req, F_ORIG);
	}

	crm_trace("Processing \"%s\" event from %s%s",
	op, originator,
	(pcmk_is_set(options, cib_force_diff)? " (global update)" : ""));

	if (*result_cib != existing_cib) {
	free_xml(*result_cib);
	}
	*result_cib = copy_xml(existing_cib);

	return xml_apply_patchset(*result_cib, input, TRUE);
	}

	// @COMPAT: v1-only
	bool
	cib__config_changed_v1(xmlNode last, xmlNode next, xmlNode **diff)
	{
	int lpc = 0, max = 0;
	bool config_changes = false;
	xmlXPathObject *xpathObj = NULL;
	int format = 1;

	CRM_ASSERT(diff != NULL);

	if (*diff == NULL && last != NULL && next != NULL) {
	*diff = diff_xml_object(last, next, FALSE);
	}

	if (*diff == NULL) {
	goto done;
	}

	crm_element_value_int(*diff, "format", &format);
	CRM_LOG_ASSERT(format == 1);

	xpathObj = xpath_search(*diff, "//" XML_CIB_TAG_CONFIGURATION);
	if (numXpathResults(xpathObj) > 0) {
	config_changes = true;
	goto done;
	}
	freeXpathObject(xpathObj);

	/*
	* Do not check XML_TAG_DIFF_ADDED "//" XML_TAG_CIB
	* This always contains every field and would produce a false positive
	* every time if the checked value existed
	*/
	xpathObj = xpath_search(*diff, "//" XML_TAG_DIFF_REMOVED "//" XML_TAG_CIB);
	max = numXpathResults(xpathObj);

	for (lpc = 0; lpc < max; lpc++) {
	xmlNode *top = getXpathResult(xpathObj, lpc);

	if (crm_element_value(top, XML_ATTR_GENERATION) != NULL) {
	config_changes = true;
	goto done;
	}
	if (crm_element_value(top, XML_ATTR_GENERATION_ADMIN) != NULL) {
	config_changes = true;
	goto done;
	}

	if (crm_element_value(top, XML_ATTR_VALIDATION) != NULL) {
	config_changes = true;
	goto done;
	}
	if (crm_element_value(top, XML_ATTR_CRM_VERSION) != NULL) {
	config_changes = true;
	goto done;
	}
	if (crm_element_value(top, "remote-clear-port") != NULL) {
	config_changes = true;
	goto done;
	}
	if (crm_element_value(top, "remote-tls-port") != NULL) {
	config_changes = true;
	goto done;
	}
	}

	done:
	freeXpathObject(xpathObj);
	return config_changes;
	}

	int
	cib_process_xpath(const char op, int options, const char section,
	const xmlNode req, xmlNode input, xmlNode *existing_cib,
	xmlNode result_cib, xmlNode answer)
	{
	int lpc = 0;
	int max = 0;
	int rc = pcmk_ok;
	bool is_query = pcmk__str_eq(op, PCMK__CIB_REQUEST_QUERY, pcmk__str_none);

	xmlXPathObjectPtr xpathObj = NULL;

	crm_trace("Processing \"%s\" event", op);

	if (is_query) {
	xpathObj = xpath_search(existing_cib, section);
	} else {
	xpathObj = xpath_search(*result_cib, section);
	}

	max = numXpathResults(xpathObj);

	if ((max < 1)
	&& pcmk__str_eq(op, PCMK__CIB_REQUEST_DELETE, pcmk__str_none)) {
	crm_debug("%s was already removed", section);

	} else if (max < 1) {
	crm_debug("%s: %s does not exist", op, section);
	rc = -ENXIO;

	} else if (is_query) {
	if (max > 1) {
	*answer = create_xml_node(NULL, "xpath-query");
	}
	}

	if (pcmk_is_set(options, cib_multiple)
	&& pcmk__str_eq(op, PCMK__CIB_REQUEST_DELETE, pcmk__str_none)) {
	dedupXpathResults(xpathObj);
	}

	for (lpc = 0; lpc < max; lpc++) {
	xmlChar *path = NULL;
	xmlNode *match = getXpathResult(xpathObj, lpc);

	if (match == NULL) {
	continue;
	}

	path = xmlGetNodePath(match);
	crm_debug("Processing %s op for %s with %s", op, section, path);
	free(path);

	if (pcmk__str_eq(op, PCMK__CIB_REQUEST_DELETE, pcmk__str_none)) {
	if (match == *result_cib) {
	/* Attempting to delete the whole "/cib" */
	crm_warn("Cannot perform %s for %s: The xpath is addressing the whole /cib", op, section);
	rc = -EINVAL;
	break;
	}

	free_xml(match);
	if ((options & cib_multiple) == 0) {
	break;
	}

	} else if (pcmk__str_eq(op, PCMK__CIB_REQUEST_MODIFY, pcmk__str_none)) {
	if (update_xml_child(match, input) == FALSE) {
	rc = -ENXIO;
	} else if ((options & cib_multiple) == 0) {
	break;
	}

	} else if (pcmk__str_eq(op, PCMK__CIB_REQUEST_CREATE, pcmk__str_none)) {
	add_node_copy(match, input);
	break;

	} else if (pcmk__str_eq(op, PCMK__CIB_REQUEST_QUERY, pcmk__str_none)) {

	if (options & cib_no_children) {
	const char *tag = TYPE(match);
	xmlNode shallow = create_xml_node(answer, tag);

	copy_in_properties(shallow, match);

	if (*answer == NULL) {
	*answer = shallow;
	}

	} else if (options & cib_xpath_address) {
	char *path = NULL;
	xmlNode *parent = match;

	while (parent && parent->type == XML_ELEMENT_NODE) {
	const char *id = crm_element_value(parent, XML_ATTR_ID);
	char *new_path = NULL;

	if (id) {
	new_path = crm_strdup_printf("/%s[@" XML_ATTR_ID "='%s']"
	"%s",
	parent->name, id,
	pcmk__s(path, ""));
	} else {
	new_path = crm_strdup_printf("/%s%s", parent->name,
	pcmk__s(path, ""));
	}
	free(path);
	path = new_path;
	parent = parent->parent;
	}
	crm_trace("Got: %s", path);

	if (*answer == NULL) {
	*answer = create_xml_node(NULL, "xpath-query");
	}
	parent = create_xml_node(*answer, "xpath-query-path");
	crm_xml_add(parent, XML_ATTR_ID, path);
	free(path);

	} else if (*answer) {
	add_node_copy(*answer, match);

	} else {
	*answer = match;
	}

	} else if (pcmk__str_eq(op, PCMK__CIB_REQUEST_REPLACE,
	pcmk__str_none)) {
	xmlNode *parent = match->parent;

	free_xml(match);
	if (input != NULL) {
	add_node_copy(parent, input);
	}

	if ((options & cib_multiple) == 0) {
	break;
	}
	}
	}

	freeXpathObject(xpathObj);
	return rc;
	}

File Metadata

Mime Type: text/x-diff
Expires: Mon, Apr 21, 12:10 PM (18 h, 49 m)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1664850
Default Alt Text: (181 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions