diff --git a/daemons/controld/controld_join_dc.c b/daemons/controld/controld_join_dc.c
index 63261399cd..b3d9526db6 100644
--- a/daemons/controld/controld_join_dc.c
+++ b/daemons/controld/controld_join_dc.c
@@ -1,1044 +1,1045 @@
 /*
  * Copyright 2004-2024 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 <crm/crm.h>
 
 #include <crm/common/xml.h>
 #include <crm/cluster.h>
 
 #include <pacemaker-controld.h>
 
 static char *max_generation_from = NULL;
 static xmlNodePtr max_generation_xml = NULL;
 
 /*!
  * \internal
  * \brief Nodes from which a CIB sync has failed since the peer joined
  *
  * This table is of the form (<tt>node_name -> join_id</tt>). \p node_name is
  * the name of a client node from which a CIB \p sync_from() call has failed in
  * \p do_dc_join_finalize() since the client joined the cluster as a peer.
  * \p join_id is the ID of the join round in which the \p sync_from() failed,
  * and is intended for use in nack log messages.
  */
 static GHashTable *failed_sync_nodes = NULL;
 
 void finalize_join_for(gpointer key, gpointer value, gpointer user_data);
 void finalize_sync_callback(xmlNode * msg, int call_id, int rc, xmlNode * output, void *user_data);
 gboolean check_join_state(enum crmd_fsa_state cur_state, const char *source);
 
 /* Numeric counter used to identify join rounds (an unsigned int would be
  * appropriate, except we get and set it in XML as int)
  */
 static int current_join_id = 0;
 
 /*!
  * \internal
  * \brief Destroy the hash table containing failed sync nodes
  */
 void
 controld_destroy_failed_sync_table(void)
 {
     if (failed_sync_nodes != NULL) {
         g_hash_table_destroy(failed_sync_nodes);
         failed_sync_nodes = NULL;
     }
 }
 
 /*!
  * \internal
  * \brief Remove a node from the failed sync nodes table if present
  *
  * \param[in] node_name  Node name to remove
  */
 void
 controld_remove_failed_sync_node(const char *node_name)
 {
     if (failed_sync_nodes != NULL) {
         g_hash_table_remove(failed_sync_nodes, (gchar *) node_name);
     }
 }
 
 /*!
  * \internal
  * \brief Add to a hash table a node whose CIB failed to sync
  *
  * \param[in] node_name  Name of node whose CIB failed to sync
  * \param[in] join_id    Join round when the failure occurred
  */
 static void
 record_failed_sync_node(const char *node_name, gint join_id)
 {
     if (failed_sync_nodes == NULL) {
         failed_sync_nodes = pcmk__strikey_table(g_free, NULL);
     }
 
     /* If the node is already in the table then we failed to nack it during the
      * filter offer step
      */
     CRM_LOG_ASSERT(g_hash_table_insert(failed_sync_nodes, g_strdup(node_name),
                                        GINT_TO_POINTER(join_id)));
 }
 
 /*!
  * \internal
  * \brief Look up a node name in the failed sync table
  *
  * \param[in]  node_name  Name of node to look up
  * \param[out] join_id    Where to store the join ID of when the sync failed
  *
  * \return Standard Pacemaker return code. Specifically, \p pcmk_rc_ok if the
  *         node name was found, or \p pcmk_rc_node_unknown otherwise.
  * \note \p *join_id is set to -1 if the node is not found.
  */
 static int
 lookup_failed_sync_node(const char *node_name, gint *join_id)
 {
     *join_id = -1;
 
     if (failed_sync_nodes != NULL) {
         gpointer result = g_hash_table_lookup(failed_sync_nodes,
                                               (gchar *) node_name);
         if (result != NULL) {
             *join_id = GPOINTER_TO_INT(result);
             return pcmk_rc_ok;
         }
     }
     return pcmk_rc_node_unknown;
 }
 
 void
 crm_update_peer_join(const char *source, crm_node_t * node, enum crm_join_phase phase)
 {
     enum crm_join_phase last = 0;
 
     CRM_CHECK(node != NULL, return);
 
     /* Remote nodes do not participate in joins */
     if (pcmk_is_set(node->flags, crm_remote_node)) {
         return;
     }
 
     last = node->join;
 
     if(phase == last) {
         crm_trace("Node %s join-%d phase is still %s "
                   CRM_XS " nodeid=%u source=%s",
                   node->uname, current_join_id, crm_join_phase_str(last),
                   node->id, source);
 
     } else if ((phase <= crm_join_none) || (phase == (last + 1))) {
         node->join = phase;
         crm_trace("Node %s join-%d phase is now %s (was %s) "
                   CRM_XS " nodeid=%u source=%s",
                  node->uname, current_join_id, crm_join_phase_str(phase),
                  crm_join_phase_str(last), node->id, source);
 
     } else {
         crm_warn("Rejecting join-%d phase update for node %s because "
                  "can't go from %s to %s " CRM_XS " nodeid=%u source=%s",
                  current_join_id, node->uname, crm_join_phase_str(last),
                  crm_join_phase_str(phase), node->id, source);
     }
 }
 
 static void
 start_join_round(void)
 {
     GHashTableIter iter;
     crm_node_t *peer = NULL;
 
     crm_debug("Starting new join round join-%d", current_join_id);
 
     g_hash_table_iter_init(&iter, crm_peer_cache);
     while (g_hash_table_iter_next(&iter, NULL, (gpointer *) &peer)) {
         crm_update_peer_join(__func__, peer, crm_join_none);
     }
     if (max_generation_from != NULL) {
         free(max_generation_from);
         max_generation_from = NULL;
     }
     if (max_generation_xml != NULL) {
         free_xml(max_generation_xml);
         max_generation_xml = NULL;
     }
     controld_clear_fsa_input_flags(R_HAVE_CIB);
 }
 
 /*!
  * \internal
  * \brief Create a join message from the DC
  *
  * \param[in] join_op  Join operation name
  * \param[in] host_to  Recipient of message
  */
 static xmlNode *
 create_dc_message(const char *join_op, const char *host_to)
 {
     xmlNode *msg = create_request(join_op, NULL, host_to, CRM_SYSTEM_CRMD,
                                   CRM_SYSTEM_DC, NULL);
 
     /* Identify which election this is a part of */
     crm_xml_add_int(msg, PCMK__XA_JOIN_ID, current_join_id);
 
     /* Add a field specifying whether the DC is shutting down. This keeps the
      * joining node from fencing the old DC if it becomes the new DC.
      */
     pcmk__xe_set_bool_attr(msg, PCMK__XA_DC_LEAVING,
                            pcmk_is_set(controld_globals.fsa_input_register,
                                        R_SHUTDOWN));
     return msg;
 }
 
 static void
 join_make_offer(gpointer key, gpointer value, gpointer user_data)
 {
     xmlNode *offer = NULL;
     crm_node_t *member = (crm_node_t *)value;
 
     CRM_ASSERT(member != NULL);
     if (crm_is_peer_active(member) == FALSE) {
         crm_info("Not making join-%d offer to inactive node %s",
                  current_join_id,
                  (member->uname? member->uname : "with unknown name"));
         if(member->expected == NULL && pcmk__str_eq(member->state, CRM_NODE_LOST, pcmk__str_casei)) {
             /* You would think this unsafe, but in fact this plus an
              * active resource is what causes it to be fenced.
              *
              * Yes, this does mean that any node that dies at the same
              * time as the old DC and is not running resource (still)
              * won't be fenced.
              *
              * I'm not happy about this either.
              */
             pcmk__update_peer_expected(__func__, member, CRMD_JOINSTATE_DOWN);
         }
         return;
     }
 
     if (member->uname == NULL) {
         crm_info("Not making join-%d offer to node uuid %s with unknown name",
                  current_join_id, member->uuid);
         return;
     }
 
     if (controld_globals.membership_id != crm_peer_seq) {
         controld_globals.membership_id = crm_peer_seq;
         crm_info("Making join-%d offers based on membership event %llu",
                  current_join_id, crm_peer_seq);
     }
 
     if(user_data && member->join > crm_join_none) {
         crm_info("Not making join-%d offer to already known node %s (%s)",
                  current_join_id, member->uname,
                  crm_join_phase_str(member->join));
         return;
     }
 
     crm_update_peer_join(__func__, (crm_node_t*)member, crm_join_none);
 
     offer = create_dc_message(CRM_OP_JOIN_OFFER, member->uname);
 
     // Advertise our feature set so the joining node can bail if not compatible
     crm_xml_add(offer, PCMK_XA_CRM_FEATURE_SET, CRM_FEATURE_SET);
 
     crm_info("Sending join-%d offer to %s", current_join_id, member->uname);
     send_cluster_message(member, crm_msg_crmd, offer, TRUE);
     free_xml(offer);
 
     crm_update_peer_join(__func__, member, crm_join_welcomed);
 }
 
 /*	 A_DC_JOIN_OFFER_ALL	*/
 void
 do_dc_join_offer_all(long long action,
                      enum crmd_fsa_cause cause,
                      enum crmd_fsa_state cur_state,
                      enum crmd_fsa_input current_input, fsa_data_t * msg_data)
 {
     int count;
 
     /* Reset everyone's status back to down or in_ccm in the CIB.
      * Any nodes that are active in the CIB but not in the cluster membership
      * will be seen as offline by the scheduler anyway.
      */
     current_join_id++;
     start_join_round();
 
     update_dc(NULL);
     if (cause == C_HA_MESSAGE && current_input == I_NODE_JOIN) {
         crm_info("A new node joined the cluster");
     }
     g_hash_table_foreach(crm_peer_cache, join_make_offer, NULL);
 
     count = crmd_join_phase_count(crm_join_welcomed);
     crm_info("Waiting on join-%d requests from %d outstanding node%s",
              current_join_id, count, pcmk__plural_s(count));
 
     // Don't waste time by invoking the scheduler yet
 }
 
 /*	 A_DC_JOIN_OFFER_ONE	*/
 void
 do_dc_join_offer_one(long long action,
                      enum crmd_fsa_cause cause,
                      enum crmd_fsa_state cur_state,
                      enum crmd_fsa_input current_input, fsa_data_t * msg_data)
 {
     crm_node_t *member;
     ha_msg_input_t *welcome = NULL;
     int count;
     const char *join_to = NULL;
 
     if (msg_data->data == NULL) {
         crm_info("Making join-%d offers to any unconfirmed nodes "
                  "because an unknown node joined", current_join_id);
         g_hash_table_foreach(crm_peer_cache, join_make_offer, &member);
         check_join_state(cur_state, __func__);
         return;
     }
 
     welcome = fsa_typed_data(fsa_dt_ha_msg);
     if (welcome == NULL) {
         // fsa_typed_data() already logged an error
         return;
     }
 
     join_to = crm_element_value(welcome->msg, PCMK__XA_SRC);
     if (join_to == NULL) {
         crm_err("Can't make join-%d offer to unknown node", current_join_id);
         return;
     }
     member = pcmk__get_node(0, join_to, NULL, pcmk__node_search_cluster);
 
     /* It is possible that a node will have been sick or starting up when the
      * original offer was made. However, it will either re-announce itself in
      * due course, or we can re-store the original offer on the client.
      */
 
     crm_update_peer_join(__func__, member, crm_join_none);
     join_make_offer(NULL, member, NULL);
 
     /* If the offer isn't to the local node, make an offer to the local node as
      * well, to ensure the correct value for max_generation_from.
      */
     if (strcasecmp(join_to, controld_globals.our_nodename) != 0) {
         member = pcmk__get_node(0, controld_globals.our_nodename, NULL,
                                 pcmk__node_search_cluster);
         join_make_offer(NULL, member, NULL);
     }
 
     /* This was a genuine join request; cancel any existing transition and
      * invoke the scheduler.
      */
     abort_transition(PCMK_SCORE_INFINITY, pcmk__graph_restart, "Node join",
                      NULL);
 
     count = crmd_join_phase_count(crm_join_welcomed);
     crm_info("Waiting on join-%d requests from %d outstanding node%s",
              current_join_id, count, pcmk__plural_s(count));
 
     // Don't waste time by invoking the scheduler yet
 }
 
 static int
 compare_int_fields(xmlNode * left, xmlNode * right, const char *field)
 {
     const char *elem_l = crm_element_value(left, field);
     const char *elem_r = crm_element_value(right, field);
 
     long long int_elem_l;
     long long int_elem_r;
 
     pcmk__scan_ll(elem_l, &int_elem_l, -1LL);
     pcmk__scan_ll(elem_r, &int_elem_r, -1LL);
 
     if (int_elem_l < int_elem_r) {
         return -1;
 
     } else if (int_elem_l > int_elem_r) {
         return 1;
     }
 
     return 0;
 }
 
 /*	 A_DC_JOIN_PROCESS_REQ	*/
 void
 do_dc_join_filter_offer(long long action,
                         enum crmd_fsa_cause cause,
                         enum crmd_fsa_state cur_state,
                         enum crmd_fsa_input current_input, fsa_data_t * msg_data)
 {
     xmlNode *generation = NULL;
 
     int cmp = 0;
     int join_id = -1;
     int count = 0;
     gint value = 0;
     gboolean ack_nack_bool = TRUE;
     ha_msg_input_t *join_ack = fsa_typed_data(fsa_dt_ha_msg);
 
     const char *join_from = crm_element_value(join_ack->msg, PCMK__XA_SRC);
     const char *ref = crm_element_value(join_ack->msg, PCMK_XA_REFERENCE);
     const char *join_version = crm_element_value(join_ack->msg,
                                                  PCMK_XA_CRM_FEATURE_SET);
     crm_node_t *join_node = NULL;
 
     if (join_from == NULL) {
         crm_err("Ignoring invalid join request without node name");
         return;
     }
     join_node = pcmk__get_node(0, join_from, NULL, pcmk__node_search_cluster);
 
     crm_element_value_int(join_ack->msg, PCMK__XA_JOIN_ID, &join_id);
     if (join_id != current_join_id) {
         crm_debug("Ignoring join-%d request from %s because we are on join-%d",
                   join_id, join_from, current_join_id);
         check_join_state(cur_state, __func__);
         return;
     }
 
     generation = join_ack->xml;
     if (max_generation_xml != NULL && generation != NULL) {
         int lpc = 0;
 
         const char *attributes[] = {
             PCMK_XA_ADMIN_EPOCH,
             PCMK_XA_EPOCH,
             PCMK_XA_NUM_UPDATES,
         };
 
         /* It's not obvious that join_ack->xml is the PCMK__XE_GENERATION_TUPLE
          * element from the join client. The "if" guard is for clarity.
          */
         if (pcmk__xe_is(generation, PCMK__XE_GENERATION_TUPLE)) {
             for (lpc = 0; cmp == 0 && lpc < PCMK__NELEM(attributes); lpc++) {
                 cmp = compare_int_fields(max_generation_xml, generation,
                                          attributes[lpc]);
             }
 
         } else {    // Should always be PCMK__XE_GENERATION_TUPLE
             CRM_LOG_ASSERT(false);
         }
     }
 
     if (ref == NULL) {
         ref = "none"; // for logging only
     }
 
     if (lookup_failed_sync_node(join_from, &value) == pcmk_rc_ok) {
         crm_err("Rejecting join-%d request from node %s because we failed to "
                 "sync its CIB in join-%d " CRM_XS " ref=%s",
                 join_id, join_from, value, ref);
         ack_nack_bool = FALSE;
 
     } else if (!crm_is_peer_active(join_node)) {
         if (match_down_event(join_from) != NULL) {
             /* The join request was received after the node was fenced or
              * otherwise shutdown in a way that we're aware of. No need to log
              * an error in this rare occurrence; we know the client was recently
              * shut down, and receiving a lingering in-flight request is not
              * cause for alarm.
              */
             crm_debug("Rejecting join-%d request from inactive node %s "
                       CRM_XS " ref=%s", join_id, join_from, ref);
         } else {
             crm_err("Rejecting join-%d request from inactive node %s "
                     CRM_XS " ref=%s", join_id, join_from, ref);
         }
         ack_nack_bool = FALSE;
 
     } else if (generation == NULL) {
         crm_err("Rejecting invalid join-%d request from node %s "
                 "missing CIB generation " CRM_XS " ref=%s",
                 join_id, join_from, ref);
         ack_nack_bool = FALSE;
 
     } else if ((join_version == NULL)
                || !feature_set_compatible(CRM_FEATURE_SET, join_version)) {
         crm_err("Rejecting join-%d request from node %s because feature set %s"
                 " is incompatible with ours (%s) " CRM_XS " ref=%s",
                 join_id, join_from, (join_version? join_version : "pre-3.1.0"),
                 CRM_FEATURE_SET, ref);
         ack_nack_bool = FALSE;
 
     } else if (max_generation_xml == NULL) {
         const char *validation = crm_element_value(generation,
                                                    PCMK_XA_VALIDATE_WITH);
 
-        if (get_schema_version(validation) < 0) {
+        if (pcmk__get_schema(validation) == NULL) {
             crm_err("Rejecting join-%d request from %s (with first CIB "
                     "generation) due to unknown schema version %s "
                     CRM_XS " ref=%s",
-                    join_id, join_from, validation, ref);
+                    join_id, join_from, pcmk__s(validation, "(missing)"), ref);
             ack_nack_bool = FALSE;
 
         } else {
             crm_debug("Accepting join-%d request from %s (with first CIB "
                       "generation) " CRM_XS " ref=%s",
                       join_id, join_from, ref);
             max_generation_xml = pcmk__xml_copy(NULL, generation);
             pcmk__str_update(&max_generation_from, join_from);
         }
 
     } else if ((cmp < 0)
                || ((cmp == 0)
                    && pcmk__str_eq(join_from, controld_globals.our_nodename,
                                    pcmk__str_casei))) {
         const char *validation = crm_element_value(generation,
                                                    PCMK_XA_VALIDATE_WITH);
 
-        if (get_schema_version(validation) < 0) {
+        if (pcmk__get_schema(validation) == NULL) {
             crm_err("Rejecting join-%d request from %s (with better CIB "
                     "generation than current best from %s) due to unknown "
                     "schema version %s " CRM_XS " ref=%s",
-                    join_id, join_from, max_generation_from, validation, ref);
+                    join_id, join_from, max_generation_from,
+                    pcmk__s(validation, "(missing)"), ref);
             ack_nack_bool = FALSE;
 
         } else {
             crm_debug("Accepting join-%d request from %s (with better CIB "
                       "generation than current best from %s) " CRM_XS " ref=%s",
                       join_id, join_from, max_generation_from, ref);
             crm_log_xml_debug(max_generation_xml, "Old max generation");
             crm_log_xml_debug(generation, "New max generation");
 
             free_xml(max_generation_xml);
             max_generation_xml = pcmk__xml_copy(NULL, join_ack->xml);
             pcmk__str_update(&max_generation_from, join_from);
         }
 
     } else {
         crm_debug("Accepting join-%d request from %s " CRM_XS " ref=%s",
                   join_id, join_from, ref);
     }
 
     if (!ack_nack_bool) {
         if (compare_version(join_version, "3.17.0") < 0) {
             /* Clients with CRM_FEATURE_SET < 3.17.0 may respawn infinitely
              * after a nack message, don't send one
              */
             crm_update_peer_join(__func__, join_node, crm_join_nack_quiet);
         } else {
             crm_update_peer_join(__func__, join_node, crm_join_nack);
         }
         pcmk__update_peer_expected(__func__, join_node, CRMD_JOINSTATE_NACK);
 
     } else {
         crm_update_peer_join(__func__, join_node, crm_join_integrated);
         pcmk__update_peer_expected(__func__, join_node, CRMD_JOINSTATE_MEMBER);
     }
 
     count = crmd_join_phase_count(crm_join_integrated);
     crm_debug("%d node%s currently integrated in join-%d",
               count, pcmk__plural_s(count), join_id);
 
     if (check_join_state(cur_state, __func__) == FALSE) {
         // Don't waste time by invoking the scheduler yet
         count = crmd_join_phase_count(crm_join_welcomed);
         crm_debug("Waiting on join-%d requests from %d outstanding node%s",
                   join_id, count, pcmk__plural_s(count));
     }
 }
 
 /*	A_DC_JOIN_FINALIZE	*/
 void
 do_dc_join_finalize(long long action,
                     enum crmd_fsa_cause cause,
                     enum crmd_fsa_state cur_state,
                     enum crmd_fsa_input current_input, fsa_data_t * msg_data)
 {
     char *sync_from = NULL;
     int rc = pcmk_ok;
     int count_welcomed = crmd_join_phase_count(crm_join_welcomed);
     int count_finalizable = crmd_join_phase_count(crm_join_integrated)
                             + crmd_join_phase_count(crm_join_nack)
                             + crmd_join_phase_count(crm_join_nack_quiet);
 
     /* This we can do straight away and avoid clients timing us out
      *  while we compute the latest CIB
      */
     if (count_welcomed != 0) {
         crm_debug("Waiting on join-%d requests from %d outstanding node%s "
                   "before finalizing join", current_join_id, count_welcomed,
                   pcmk__plural_s(count_welcomed));
         crmd_join_phase_log(LOG_DEBUG);
         /* crmd_fsa_stall(FALSE); Needed? */
         return;
 
     } else if (count_finalizable == 0) {
         crm_debug("Finalization not needed for join-%d at the current time",
                   current_join_id);
         crmd_join_phase_log(LOG_DEBUG);
         check_join_state(controld_globals.fsa_state, __func__);
         return;
     }
 
     controld_clear_fsa_input_flags(R_HAVE_CIB);
     if (pcmk__str_eq(max_generation_from, controld_globals.our_nodename,
                      pcmk__str_null_matches|pcmk__str_casei)) {
         controld_set_fsa_input_flags(R_HAVE_CIB);
     }
 
     if (!controld_globals.transition_graph->complete) {
         crm_warn("Delaying join-%d finalization while transition in progress",
                  current_join_id);
         crmd_join_phase_log(LOG_DEBUG);
         crmd_fsa_stall(FALSE);
         return;
     }
 
     if (pcmk_is_set(controld_globals.fsa_input_register, R_HAVE_CIB)) {
         // Send our CIB out to everyone
         sync_from = pcmk__str_copy(controld_globals.our_nodename);
         crm_debug("Finalizing join-%d for %d node%s (sync'ing from local CIB)",
                   current_join_id, count_finalizable,
                   pcmk__plural_s(count_finalizable));
         crm_log_xml_debug(max_generation_xml, "Requested CIB version");
 
     } else {
         // Ask for the agreed best CIB
         sync_from = pcmk__str_copy(max_generation_from);
         crm_notice("Finalizing join-%d for %d node%s (sync'ing CIB from %s)",
                    current_join_id, count_finalizable,
                    pcmk__plural_s(count_finalizable), sync_from);
         crm_log_xml_notice(max_generation_xml, "Requested CIB version");
     }
     crmd_join_phase_log(LOG_DEBUG);
 
     rc = controld_globals.cib_conn->cmds->sync_from(controld_globals.cib_conn,
                                                     sync_from, NULL, cib_none);
     fsa_register_cib_callback(rc, sync_from, finalize_sync_callback);
 }
 
 void
 free_max_generation(void)
 {
     free(max_generation_from);
     max_generation_from = NULL;
 
     free_xml(max_generation_xml);
     max_generation_xml = NULL;
 }
 
 void
 finalize_sync_callback(xmlNode * msg, int call_id, int rc, xmlNode * output, void *user_data)
 {
     CRM_LOG_ASSERT(-EPERM != rc);
 
     if (rc != pcmk_ok) {
         const char *sync_from = (const char *) user_data;
 
         do_crm_log(((rc == -pcmk_err_old_data)? LOG_WARNING : LOG_ERR),
                    "Could not sync CIB from %s in join-%d: %s",
                    sync_from, current_join_id, pcmk_strerror(rc));
 
         if (rc != -pcmk_err_old_data) {
             record_failed_sync_node(sync_from, current_join_id);
         }
 
         /* restart the whole join process */
         register_fsa_error_adv(C_FSA_INTERNAL, I_ELECTION_DC, NULL, NULL,
                                __func__);
 
     } else if (!AM_I_DC) {
         crm_debug("Sync'ed CIB for join-%d but no longer DC", current_join_id);
 
     } else if (controld_globals.fsa_state != S_FINALIZE_JOIN) {
         crm_debug("Sync'ed CIB for join-%d but no longer in S_FINALIZE_JOIN "
                   "(%s)", current_join_id,
                   fsa_state2string(controld_globals.fsa_state));
 
     } else {
         controld_set_fsa_input_flags(R_HAVE_CIB);
 
         /* make sure dc_uuid is re-set to us */
         if (!check_join_state(controld_globals.fsa_state, __func__)) {
             int count_finalizable = 0;
 
             count_finalizable = crmd_join_phase_count(crm_join_integrated)
                                 + crmd_join_phase_count(crm_join_nack)
                                 + crmd_join_phase_count(crm_join_nack_quiet);
 
             crm_debug("Notifying %d node%s of join-%d results",
                       count_finalizable, pcmk__plural_s(count_finalizable),
                       current_join_id);
             g_hash_table_foreach(crm_peer_cache, finalize_join_for, NULL);
         }
     }
 }
 
 static void
 join_node_state_commit_callback(xmlNode *msg, int call_id, int rc,
                                 xmlNode *output, void *user_data)
 {
     const char *node = user_data;
 
     if (rc != pcmk_ok) {
         fsa_data_t *msg_data = NULL;    // for register_fsa_error() macro
 
         crm_crit("join-%d node history update (via CIB call %d) for node %s "
                  "failed: %s",
                  current_join_id, call_id, node, pcmk_strerror(rc));
         crm_log_xml_debug(msg, "failed");
         register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL);
     }
 
     crm_debug("join-%d node history update (via CIB call %d) for node %s "
               "complete",
               current_join_id, call_id, node);
     check_join_state(controld_globals.fsa_state, __func__);
 }
 
 /*	A_DC_JOIN_PROCESS_ACK	*/
 void
 do_dc_join_ack(long long action,
                enum crmd_fsa_cause cause,
                enum crmd_fsa_state cur_state,
                enum crmd_fsa_input current_input, fsa_data_t * msg_data)
 {
     int join_id = -1;
     ha_msg_input_t *join_ack = fsa_typed_data(fsa_dt_ha_msg);
 
     const char *op = crm_element_value(join_ack->msg, PCMK__XA_CRM_TASK);
     char *join_from = crm_element_value_copy(join_ack->msg, PCMK__XA_SRC);
     crm_node_t *peer = NULL;
 
     enum controld_section_e section = controld_section_lrm;
     char *xpath = NULL;
     xmlNode *state = join_ack->xml;
     xmlNode *execd_state = NULL;
 
     cib_t *cib = controld_globals.cib_conn;
     int rc = pcmk_ok;
 
     // Sanity checks
     if (join_from == NULL) {
         crm_warn("Ignoring message received without node identification");
         goto done;
     }
     if (op == NULL) {
         crm_warn("Ignoring message received from %s without task", join_from);
         goto done;
     }
 
     if (strcmp(op, CRM_OP_JOIN_CONFIRM)) {
         crm_debug("Ignoring '%s' message from %s while waiting for '%s'",
                   op, join_from, CRM_OP_JOIN_CONFIRM);
         goto done;
     }
 
     if (crm_element_value_int(join_ack->msg, PCMK__XA_JOIN_ID, &join_id) != 0) {
         crm_warn("Ignoring join confirmation from %s without valid join ID",
                  join_from);
         goto done;
     }
 
     peer = pcmk__get_node(0, join_from, NULL, pcmk__node_search_cluster);
     if (peer->join != crm_join_finalized) {
         crm_info("Ignoring out-of-sequence join-%d confirmation from %s "
                  "(currently %s not %s)",
                  join_id, join_from, crm_join_phase_str(peer->join),
                  crm_join_phase_str(crm_join_finalized));
         goto done;
     }
 
     if (join_id != current_join_id) {
         crm_err("Rejecting join-%d confirmation from %s "
                 "because currently on join-%d",
                 join_id, join_from, current_join_id);
         crm_update_peer_join(__func__, peer, crm_join_nack);
         goto done;
     }
 
     crm_update_peer_join(__func__, peer, crm_join_confirmed);
 
     /* Update CIB with node's current executor state. A new transition will be
      * triggered later, when the CIB manager notifies us of the change.
      *
      * The delete and modify requests are part of an atomic transaction.
      */
     rc = cib->cmds->init_transaction(cib);
     if (rc != pcmk_ok) {
         goto done;
     }
 
     // Delete relevant parts of node's current executor state from CIB
     if (pcmk_is_set(controld_globals.flags, controld_shutdown_lock_enabled)) {
         section = controld_section_lrm_unlocked;
     }
     controld_node_state_deletion_strings(join_from, section, &xpath, NULL);
 
     rc = cib->cmds->remove(cib, xpath, NULL,
                            cib_scope_local
                            |cib_xpath
                            |cib_multiple
                            |cib_transaction);
     if (rc != pcmk_ok) {
         goto done;
     }
 
     // Update CIB with node's latest known executor state
     if (pcmk__str_eq(join_from, controld_globals.our_nodename,
                      pcmk__str_casei)) {
 
         // Use the latest possible state if processing our own join ack
         execd_state = controld_query_executor_state();
 
         if (execd_state != NULL) {
             crm_debug("Updating local node history for join-%d from query "
                       "result",
                       current_join_id);
             state = execd_state;
 
         } else {
             crm_warn("Updating local node history from join-%d confirmation "
                      "because query failed",
                      current_join_id);
         }
 
     } else {
         crm_debug("Updating node history for %s from join-%d confirmation",
                   join_from, current_join_id);
     }
 
     rc = cib->cmds->modify(cib, PCMK_XE_STATUS, state,
                            cib_scope_local|cib_can_create|cib_transaction);
     free_xml(execd_state);
     if (rc != pcmk_ok) {
         goto done;
     }
 
     // Commit the transaction
     rc = cib->cmds->end_transaction(cib, true, cib_scope_local);
     fsa_register_cib_callback(rc, join_from, join_node_state_commit_callback);
 
     if (rc > 0) {
         // join_from will be freed after callback
         join_from = NULL;
         rc = pcmk_ok;
     }
 
 done:
     if (rc != pcmk_ok) {
         crm_crit("join-%d node history update for node %s failed: %s",
                  current_join_id, join_from, pcmk_strerror(rc));
         register_fsa_error(C_FSA_INTERNAL, I_ERROR, NULL);
     }
     free(join_from);
     free(xpath);
 }
 
 void
 finalize_join_for(gpointer key, gpointer value, gpointer user_data)
 {
     xmlNode *acknak = NULL;
     xmlNode *tmp1 = NULL;
     crm_node_t *join_node = value;
     const char *join_to = join_node->uname;
     bool integrated = false;
 
     switch (join_node->join) {
         case crm_join_integrated:
             integrated = true;
             break;
         case crm_join_nack:
         case crm_join_nack_quiet:
             break;
         default:
             crm_trace("Not updating non-integrated and non-nacked node %s (%s) "
                       "for join-%d", join_to,
                       crm_join_phase_str(join_node->join), current_join_id);
             return;
     }
 
     /* Update the <node> element with the node's name and UUID, in case they
      * weren't known before
      */
     crm_trace("Updating node name and UUID in CIB for %s", join_to);
     tmp1 = pcmk__xe_create(NULL, PCMK_XE_NODE);
     crm_xml_add(tmp1, PCMK_XA_ID, crm_peer_uuid(join_node));
     crm_xml_add(tmp1, PCMK_XA_UNAME, join_to);
     fsa_cib_anon_update(PCMK_XE_NODES, tmp1);
     free_xml(tmp1);
 
     if (join_node->join == crm_join_nack_quiet) {
         crm_trace("Not sending nack message to node %s with feature set older "
                   "than 3.17.0", join_to);
         return;
     }
 
     join_node = pcmk__get_node(0, join_to, NULL, pcmk__node_search_cluster);
     if (!crm_is_peer_active(join_node)) {
         /*
          * NACK'ing nodes that the membership layer doesn't know about yet
          * simply creates more churn
          *
          * Better to leave them waiting and let the join restart when
          * the new membership event comes in
          *
          * All other NACKs (due to versions etc) should still be processed
          */
         pcmk__update_peer_expected(__func__, join_node, CRMD_JOINSTATE_PENDING);
         return;
     }
 
     // Acknowledge or nack node's join request
     crm_debug("%sing join-%d request from %s",
               integrated? "Acknowledg" : "Nack", current_join_id, join_to);
     acknak = create_dc_message(CRM_OP_JOIN_ACKNAK, join_to);
     pcmk__xe_set_bool_attr(acknak, CRM_OP_JOIN_ACKNAK, integrated);
 
     if (integrated) {
         // No change needed for a nacked node
         crm_update_peer_join(__func__, join_node, crm_join_finalized);
         pcmk__update_peer_expected(__func__, join_node, CRMD_JOINSTATE_MEMBER);
 
         /* Iterate through the remote peer cache and add information on which
          * node hosts each to the ACK message.  This keeps new controllers in
          * sync with what has already happened.
          */
         if (crm_remote_peer_cache_size() != 0) {
             GHashTableIter iter;
             crm_node_t *node = NULL;
             xmlNode *remotes = pcmk__xe_create(acknak, PCMK_XE_NODES);
 
             g_hash_table_iter_init(&iter, crm_remote_peer_cache);
             while (g_hash_table_iter_next(&iter, NULL, (gpointer *) &node)) {
                 xmlNode *remote = NULL;
 
                 if (!node->conn_host) {
                     continue;
                 }
 
                 remote = pcmk__xe_create(remotes, PCMK_XE_NODE);
                 pcmk__xe_set_props(remote,
                                    PCMK_XA_ID, node->uname,
                                    PCMK__XA_NODE_STATE, node->state,
                                    PCMK__XA_CONNECTION_HOST, node->conn_host,
                                    NULL);
             }
         }
     }
     send_cluster_message(join_node, crm_msg_crmd, acknak, TRUE);
     free_xml(acknak);
     return;
 }
 
 gboolean
 check_join_state(enum crmd_fsa_state cur_state, const char *source)
 {
     static unsigned long long highest_seq = 0;
 
     if (controld_globals.membership_id != crm_peer_seq) {
         crm_debug("join-%d: Membership changed from %llu to %llu "
                   CRM_XS " highest=%llu state=%s for=%s",
                   current_join_id, controld_globals.membership_id, crm_peer_seq,
                   highest_seq, fsa_state2string(cur_state), source);
         if(highest_seq < crm_peer_seq) {
             /* Don't spam the FSA with duplicates */
             highest_seq = crm_peer_seq;
             register_fsa_input_before(C_FSA_INTERNAL, I_NODE_JOIN, NULL);
         }
 
     } else if (cur_state == S_INTEGRATION) {
         if (crmd_join_phase_count(crm_join_welcomed) == 0) {
             int count = crmd_join_phase_count(crm_join_integrated);
 
             crm_debug("join-%d: Integration of %d peer%s complete "
                       CRM_XS " state=%s for=%s",
                       current_join_id, count, pcmk__plural_s(count),
                       fsa_state2string(cur_state), source);
             register_fsa_input_before(C_FSA_INTERNAL, I_INTEGRATED, NULL);
             return TRUE;
         }
 
     } else if (cur_state == S_FINALIZE_JOIN) {
         if (!pcmk_is_set(controld_globals.fsa_input_register, R_HAVE_CIB)) {
             crm_debug("join-%d: Delaying finalization until we have CIB "
                       CRM_XS " state=%s for=%s",
                       current_join_id, fsa_state2string(cur_state), source);
             return TRUE;
 
         } else if (crmd_join_phase_count(crm_join_welcomed) != 0) {
             int count = crmd_join_phase_count(crm_join_welcomed);
 
             crm_debug("join-%d: Still waiting on %d welcomed node%s "
                       CRM_XS " state=%s for=%s",
                       current_join_id, count, pcmk__plural_s(count),
                       fsa_state2string(cur_state), source);
             crmd_join_phase_log(LOG_DEBUG);
 
         } else if (crmd_join_phase_count(crm_join_integrated) != 0) {
             int count = crmd_join_phase_count(crm_join_integrated);
 
             crm_debug("join-%d: Still waiting on %d integrated node%s "
                       CRM_XS " state=%s for=%s",
                       current_join_id, count, pcmk__plural_s(count),
                       fsa_state2string(cur_state), source);
             crmd_join_phase_log(LOG_DEBUG);
 
         } else if (crmd_join_phase_count(crm_join_finalized) != 0) {
             int count = crmd_join_phase_count(crm_join_finalized);
 
             crm_debug("join-%d: Still waiting on %d finalized node%s "
                       CRM_XS " state=%s for=%s",
                       current_join_id, count, pcmk__plural_s(count),
                       fsa_state2string(cur_state), source);
             crmd_join_phase_log(LOG_DEBUG);
 
         } else {
             crm_debug("join-%d: Complete " CRM_XS " state=%s for=%s",
                       current_join_id, fsa_state2string(cur_state), source);
             register_fsa_input_later(C_FSA_INTERNAL, I_FINALIZED, NULL);
             return TRUE;
         }
     }
 
     return FALSE;
 }
 
 void
 do_dc_join_final(long long action,
                  enum crmd_fsa_cause cause,
                  enum crmd_fsa_state cur_state,
                  enum crmd_fsa_input current_input, fsa_data_t * msg_data)
 {
     crm_debug("Ensuring DC, quorum and node attributes are up-to-date");
     crm_update_quorum(crm_have_quorum, TRUE);
 }
 
 int crmd_join_phase_count(enum crm_join_phase phase)
 {
     int count = 0;
     crm_node_t *peer;
     GHashTableIter iter;
 
     g_hash_table_iter_init(&iter, crm_peer_cache);
     while (g_hash_table_iter_next(&iter, NULL, (gpointer *) &peer)) {
         if(peer->join == phase) {
             count++;
         }
     }
     return count;
 }
 
 void crmd_join_phase_log(int level)
 {
     crm_node_t *peer;
     GHashTableIter iter;
 
     g_hash_table_iter_init(&iter, crm_peer_cache);
     while (g_hash_table_iter_next(&iter, NULL, (gpointer *) &peer)) {
         do_crm_log(level, "join-%d: %s=%s", current_join_id, peer->uname,
                    crm_join_phase_str(peer->join));
     }
 }
diff --git a/include/crm/common/schemas_internal.h b/include/crm/common/schemas_internal.h
index c2cf0e63ed..96283f6598 100644
--- a/include/crm/common/schemas_internal.h
+++ b/include/crm/common/schemas_internal.h
@@ -1,29 +1,30 @@
 /*
  * Copyright 2006-2024 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 PCMK__SCHEMAS_INTERNAL__H
 #  define PCMK__SCHEMAS_INTERNAL__H
 
 #include <glib.h>           // GList, gboolean
 #include <libxml/tree.h>    // xmlNode, xmlRelaxNGValidityErrorFunc
 
 void crm_schema_init(void);
 void crm_schema_cleanup(void);
 
 void pcmk__load_schemas_from_dir(const char *dir);
 void pcmk__sort_schemas(void);
 GList *pcmk__schema_files_later_than(const char *name);
 void pcmk__build_schema_xml_node(xmlNode *parent, const char *name,
                                  GList **already_included);
 const char *pcmk__remote_schema_dir(void);
+GList *pcmk__get_schema(const char *name);
 gboolean pcmk__validate_xml(xmlNode *xml_blob, const char *validation,
                             xmlRelaxNGValidityErrorFunc error_handler,
                             void *error_handler_context);
 
 #endif // PCMK__SCHEMAS_INTERNAL__H
diff --git a/lib/common/schemas.c b/lib/common/schemas.c
index eb28d2ca87..30f0b72c63 100644
--- a/lib/common/schemas.c
+++ b/lib/common/schemas.c
@@ -1,1605 +1,1629 @@
 /*
  * Copyright 2004-2024 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 <string.h>
 #include <dirent.h>
 #include <errno.h>
 #include <sys/stat.h>
 #include <stdarg.h>
 
 #include <libxml/relaxng.h>
 #include <libxslt/xslt.h>
 #include <libxslt/transform.h>
 #include <libxslt/security.h>
 #include <libxslt/xsltutils.h>
 
 #include <crm/common/xml.h>
 #include <crm/common/xml_internal.h>  /* PCMK__XML_LOG_BASE */
 
 #include "crmcommon_private.h"
 
 #define SCHEMA_ZERO { .v = { 0, 0 } }
 
 #define schema_strdup_printf(prefix, version, suffix) \
     crm_strdup_printf(prefix "%u.%u" suffix, (version).v[0], (version).v[1])
 
 typedef struct {
     xmlRelaxNGPtr rng;
     xmlRelaxNGValidCtxtPtr valid;
     xmlRelaxNGParserCtxtPtr parser;
 } relaxng_ctx_cache_t;
 
 static GList *known_schemas = NULL;
 static bool silent_logging = FALSE;
 
 static void G_GNUC_PRINTF(2, 3)
 xml_log(int priority, const char *fmt, ...)
 {
     va_list ap;
 
     va_start(ap, fmt);
     if (silent_logging == FALSE) {
         /* XXX should not this enable dechunking as well? */
         PCMK__XML_LOG_BASE(priority, FALSE, 0, NULL, fmt, ap);
     }
     va_end(ap);
 }
 
 static int
 xml_latest_schema_index(void)
 {
     /* This function assumes that crm_schema_init() has been called beforehand,
      * so we have at least three schemas (one real schema, the "pacemaker-next"
      * schema, and the "none" schema).
      *
      * @COMPAT: pacemaker-next is deprecated since 2.1.5.
      * Update this when we drop that schema.
      */
     return g_list_length(known_schemas) - 3;
 }
 
 /* Return the index of the most recent X.0 schema. */
 int
 pcmk__find_x_0_schema_index(void)
 {
     /* We can't just use best to determine whether we've found the index
      * or not.  What if we have a very long list of schemas all in the
      * same major version series?  We'd return 0 for that, which means
      * we would still run this function every time.
      */
 #if defined(PCMK__UNIT_TESTING)
     /* If we're unit testing, these can't be static because they'll stick
      * around from one test run to the next.  They need to be cleared out
      * every time.
      */
     bool found = false;
     int best = 0;
 #else
     static bool found = false;
     static int best = 0;
 #endif
     int i;
     GList *best_node = NULL;
     pcmk__schema_t *best_schema = NULL;
 
     if (found) {
         return best;
     }
 
     CRM_ASSERT(known_schemas != NULL);
 
     /* Get the most recent schema so we can look at its version number. */
     best = xml_latest_schema_index();
     best_node = g_list_nth(known_schemas, best);
     best_schema = best_node->data;
 
     /* The "pacemaker-next" and "none" schemas are added to the real schemas,
      * so a list of length three actually only has one useful schema.
      *
      * @COMPAT pacemaker-next is deprecated since 2.1.5.
      * Update this when we drop that schema.
      */
     if (g_list_length(known_schemas) == 3) {
         goto done;
     }
 
     /* Start comparing the list from the node before the best schema (there's
      * no point in comparing something to itself).  Then, 'i' is an index
      * starting at the best schema and will always point at the node after
      * 'iter'.  This makes it the value we want to return when we find what
      * we're looking for.
      */
     i = best;
 
     for (GList *iter = best_node->prev; iter != NULL; iter = iter->prev) {
         pcmk__schema_t *schema = iter->data;
 
         /* We've found a schema in an older major version series.  Return
          * the index of the first one in the same major version series as
          * the best schema.
          */
         if (schema->version.v[0] < best_schema->version.v[0]) {
             best = i;
             goto done;
 
         /* We're out of list to examine.  This probably means there was only
          * one major version series, so return index 0.
          */
         } else if (iter->prev == NULL) {
             best = 0;
             goto done;
         }
 
         i--;
     }
 
 done:
     found = true;
     return best;
 }
 
 const char *
 xml_latest_schema(void)
 {
     return get_schema_name(xml_latest_schema_index());
 }
 
 static inline bool
 version_from_filename(const char *filename, pcmk__schema_version_t *version)
 {
     if (pcmk__ends_with(filename, ".rng")) {
         return sscanf(filename, "pacemaker-%hhu.%hhu.rng", &(version->v[0]), &(version->v[1])) == 2;
     } else {
         return sscanf(filename, "pacemaker-%hhu.%hhu", &(version->v[0]), &(version->v[1])) == 2;
     }
 }
 
 static int
 schema_filter(const struct dirent *a)
 {
     int rc = 0;
     pcmk__schema_version_t version = SCHEMA_ZERO;
 
     if (strstr(a->d_name, "pacemaker-") != a->d_name) {
         /* crm_trace("%s - wrong prefix", a->d_name); */
 
     } else if (!pcmk__ends_with_ext(a->d_name, ".rng")) {
         /* crm_trace("%s - wrong suffix", a->d_name); */
 
     } else if (!version_from_filename(a->d_name, &version)) {
         /* crm_trace("%s - wrong format", a->d_name); */
 
     } else {
         /* crm_debug("%s - candidate", a->d_name); */
         rc = 1;
     }
 
     return rc;
 }
 
 static int
 schema_cmp(pcmk__schema_version_t a_version, pcmk__schema_version_t b_version)
 {
     for (int i = 0; i < 2; ++i) {
         if (a_version.v[i] < b_version.v[i]) {
             return -1;
         } else if (a_version.v[i] > b_version.v[i]) {
             return 1;
         }
     }
     return 0;
 }
 
 static int
 schema_cmp_directory(const struct dirent **a, const struct dirent **b)
 {
     pcmk__schema_version_t a_version = SCHEMA_ZERO;
     pcmk__schema_version_t b_version = SCHEMA_ZERO;
 
     if (!version_from_filename(a[0]->d_name, &a_version)
         || !version_from_filename(b[0]->d_name, &b_version)) {
         // Shouldn't be possible, but makes static analysis happy
         return 0;
     }
 
     return schema_cmp(a_version, b_version);
 }
 
 /*!
  * \internal
  * \brief Add given schema + auxiliary data to internal bookkeeping.
  *
  * \note When providing \p version, should not be called directly but
  *       through \c add_schema_by_version.
  */
 static void
 add_schema(enum pcmk__schema_validator validator, const pcmk__schema_version_t *version,
            const char *name, const char *transform,
            const char *transform_enter, bool transform_onleave)
 {
     pcmk__schema_t *schema = NULL;
 
     schema = pcmk__assert_alloc(1, sizeof(pcmk__schema_t));
 
     schema->validator = validator;
     schema->version.v[0] = version->v[0];
     schema->version.v[1] = version->v[1];
     schema->transform_onleave = transform_onleave;
     // schema->schema_index is set after all schemas are loaded and sorted
 
     if (version->v[0] || version->v[1]) {
         schema->name = schema_strdup_printf("pacemaker-", *version, "");
     } else {
         schema->name = pcmk__str_copy(name);
     }
 
     if (transform) {
         schema->transform = pcmk__str_copy(transform);
     }
 
     if (transform_enter) {
         schema->transform_enter = pcmk__str_copy(transform_enter);
     }
 
     known_schemas = g_list_prepend(known_schemas, schema);
 }
 
 /*!
  * \internal
  * \brief Add version-specified schema + auxiliary data to internal bookkeeping.
  * \return Standard Pacemaker return value (the only possible values are
  * \c ENOENT when no upgrade schema is associated, or \c pcmk_rc_ok otherwise.
  *
  * \note There's no reliance on the particular order of schemas entering here.
  *
  * \par A bit of theory
  * We track 3 XSLT stylesheets that differ per usage:
  * - "upgrade":
  *   . sparsely spread over the sequence of all available schemas,
  *     as they are only relevant when major version of the schema
  *     is getting bumped -- in that case, it MUST be set
  *   . name convention:  upgrade-X.Y.xsl
  * - "upgrade-enter":
  *   . may only accompany "upgrade" occurrence, but doesn't need to
  *     be present anytime such one is, i.e., it MAY not be set when
  *     "upgrade" is
  *   . name convention:  upgrade-X.Y-enter.xsl,
  *     when not present: upgrade-enter.xsl
  * - "upgrade-leave":
  *   . like "upgrade-enter", but SHOULD be present whenever
  *     "upgrade-enter" is (and vice versa, but that's only
  *     to prevent confusion based on observing the files,
  *     it would get ignored regardless)
  *   . name convention:  (see "upgrade-enter")
  */
 static int
 add_schema_by_version(const pcmk__schema_version_t *version, bool transform_expected)
 {
     bool transform_onleave = FALSE;
     int rc = pcmk_rc_ok;
     struct stat s;
     char *xslt = NULL,
          *transform_upgrade = NULL,
          *transform_enter = NULL;
 
     /* prologue for further transform_expected handling */
     if (transform_expected) {
         /* check if there's suitable "upgrade" stylesheet */
         transform_upgrade = schema_strdup_printf("upgrade-", *version, );
         xslt = pcmk__xml_artefact_path(pcmk__xml_artefact_ns_legacy_xslt,
                                        transform_upgrade);
     }
 
     if (!transform_expected) {
         /* jump directly to the end */
 
     } else if (stat(xslt, &s) == 0) {
         /* perhaps there's also a targeted "upgrade-enter" stylesheet */
         transform_enter = schema_strdup_printf("upgrade-", *version, "-enter");
         free(xslt);
         xslt = pcmk__xml_artefact_path(pcmk__xml_artefact_ns_legacy_xslt,
                                        transform_enter);
         if (stat(xslt, &s) != 0) {
             /* or initially, at least a generic one */
             crm_debug("Upgrade-enter transform %s.xsl not found", xslt);
             free(xslt);
             free(transform_enter);
             transform_enter = strdup("upgrade-enter");
             xslt = pcmk__xml_artefact_path(pcmk__xml_artefact_ns_legacy_xslt,
                                            transform_enter);
             if (stat(xslt, &s) != 0) {
                 crm_debug("Upgrade-enter transform %s.xsl not found, either", xslt);
                 free(xslt);
                 xslt = NULL;
             }
         }
         /* xslt contains full path to "upgrade-enter" stylesheet */
         if (xslt != NULL) {
             /* then there should be "upgrade-leave" counterpart (enter->leave) */
             memcpy(strrchr(xslt, '-') + 1, "leave", sizeof("leave") - 1);
             transform_onleave = (stat(xslt, &s) == 0);
             free(xslt);
         } else {
             free(transform_enter);
             transform_enter = NULL;
         }
 
     } else {
         crm_err("Upgrade transform %s not found", xslt);
         free(xslt);
         free(transform_upgrade);
         transform_upgrade = NULL;
         rc = ENOENT;
     }
 
     add_schema(pcmk__schema_validator_rng, version, NULL,
                transform_upgrade, transform_enter, transform_onleave);
 
     free(transform_upgrade);
     free(transform_enter);
 
     return rc;
 }
 
 static void
 wrap_libxslt(bool finalize)
 {
     static xsltSecurityPrefsPtr secprefs;
     int ret = 0;
 
     /* security framework preferences */
     if (!finalize) {
         CRM_ASSERT(secprefs == NULL);
         secprefs = xsltNewSecurityPrefs();
         ret = xsltSetSecurityPrefs(secprefs, XSLT_SECPREF_WRITE_FILE,
                                    xsltSecurityForbid)
               | xsltSetSecurityPrefs(secprefs, XSLT_SECPREF_CREATE_DIRECTORY,
                                      xsltSecurityForbid)
               | xsltSetSecurityPrefs(secprefs, XSLT_SECPREF_READ_NETWORK,
                                      xsltSecurityForbid)
               | xsltSetSecurityPrefs(secprefs, XSLT_SECPREF_WRITE_NETWORK,
                                      xsltSecurityForbid);
         if (ret != 0) {
             return;
         }
     } else {
         xsltFreeSecurityPrefs(secprefs);
         secprefs = NULL;
     }
 
     /* cleanup only */
     if (finalize) {
         xsltCleanupGlobals();
     }
 }
 
 void
 pcmk__load_schemas_from_dir(const char *dir)
 {
     int lpc, max;
     struct dirent **namelist = NULL;
 
     max = scandir(dir, &namelist, schema_filter, schema_cmp_directory);
     if (max < 0) {
         crm_warn("Could not load schemas from %s: %s", dir, strerror(errno));
         return;
     }
 
     for (lpc = 0; lpc < max; lpc++) {
         bool transform_expected = false;
         pcmk__schema_version_t version = SCHEMA_ZERO;
 
         if (!version_from_filename(namelist[lpc]->d_name, &version)) {
             // Shouldn't be possible, but makes static analysis happy
             crm_warn("Skipping schema '%s': could not parse version",
                      namelist[lpc]->d_name);
             continue;
         }
         if ((lpc + 1) < max) {
             pcmk__schema_version_t next_version = SCHEMA_ZERO;
 
             if (version_from_filename(namelist[lpc+1]->d_name, &next_version)
                     && (version.v[0] < next_version.v[0])) {
                 transform_expected = true;
             }
         }
 
         if (add_schema_by_version(&version, transform_expected) != pcmk_rc_ok) {
             break;
         }
     }
 
     for (lpc = 0; lpc < max; lpc++) {
         free(namelist[lpc]);
     }
 
     free(namelist);
 }
 
 static gint
 schema_sort_GCompareFunc(gconstpointer a, gconstpointer b)
 {
     const pcmk__schema_t *schema_a = a;
     const pcmk__schema_t *schema_b = b;
 
     if (pcmk__str_eq(schema_a->name, "pacemaker-next", pcmk__str_none)) {
         if (pcmk__str_eq(schema_b->name, PCMK_VALUE_NONE, pcmk__str_none)) {
             return -1;
         } else {
             return 1;
         }
     } else if (pcmk__str_eq(schema_a->name, PCMK_VALUE_NONE, pcmk__str_none)) {
         return 1;
     } else if (pcmk__str_eq(schema_b->name, "pacemaker-next", pcmk__str_none)) {
         return -1;
     } else {
         return schema_cmp(schema_a->version, schema_b->version);
     }
 }
 
 /*!
  * \internal
  * \brief Sort the list of known schemas such that all pacemaker-X.Y are in
  *        version order, then pacemaker-next, then none
  *
  * This function should be called whenever additional schemas are loaded using
  * pcmk__load_schemas_from_dir(), after the initial sets in crm_schema_init().
  */
 void
 pcmk__sort_schemas(void)
 {
     known_schemas = g_list_sort(known_schemas, schema_sort_GCompareFunc);
 }
 
 /*!
  * \internal
  * \brief Load pacemaker schemas into cache
  *
  * \note This currently also serves as an entry point for the
  *       generic initialization of the libxslt library.
  */
 void
 crm_schema_init(void)
 {
     const char *remote_schema_dir = pcmk__remote_schema_dir();
     char *base = pcmk__xml_artefact_root(pcmk__xml_artefact_ns_legacy_rng);
     const pcmk__schema_version_t zero = SCHEMA_ZERO;
     int schema_index = 0;
 
     wrap_libxslt(false);
 
     pcmk__load_schemas_from_dir(base);
     pcmk__load_schemas_from_dir(remote_schema_dir);
 
     // @COMPAT: Deprecated since 2.1.5
     add_schema(pcmk__schema_validator_rng, &zero, "pacemaker-next",
                NULL, NULL, FALSE);
 
     add_schema(pcmk__schema_validator_none, &zero, PCMK_VALUE_NONE,
                NULL, NULL, FALSE);
 
     /* add_schema() prepends items to the list, so in the simple case, this just
      * reverses the list. However if there were any remote schemas, sorting is
      * necessary.
      */
     pcmk__sort_schemas();
 
     // Now set the schema indexes and log the final result
     for (GList *iter = known_schemas; iter != NULL; iter = iter->next) {
         pcmk__schema_t *schema = iter->data;
 
         if (schema->transform == NULL) {
             crm_debug("Loaded schema %d: %s", schema_index, schema->name);
         } else {
             crm_debug("Loaded schema %d: %s (upgrades with %s.xsl)",
                       schema_index, schema->name, schema->transform);
         }
         schema->schema_index = schema_index++;
     }
 }
 
 static gboolean
 validate_with_relaxng(xmlDocPtr doc, xmlRelaxNGValidityErrorFunc error_handler, void *error_handler_context, const char *relaxng_file,
                       relaxng_ctx_cache_t **cached_ctx)
 {
     int rc = 0;
     gboolean valid = TRUE;
     relaxng_ctx_cache_t *ctx = NULL;
 
     CRM_CHECK(doc != NULL, return FALSE);
     CRM_CHECK(relaxng_file != NULL, return FALSE);
 
     if (cached_ctx && *cached_ctx) {
         ctx = *cached_ctx;
 
     } else {
         crm_debug("Creating RNG parser context");
         ctx = pcmk__assert_alloc(1, sizeof(relaxng_ctx_cache_t));
 
         ctx->parser = xmlRelaxNGNewParserCtxt(relaxng_file);
         CRM_CHECK(ctx->parser != NULL, goto cleanup);
 
         if (error_handler) {
             xmlRelaxNGSetParserErrors(ctx->parser,
                                       (xmlRelaxNGValidityErrorFunc) error_handler,
                                       (xmlRelaxNGValidityWarningFunc) error_handler,
                                       error_handler_context);
         } else {
             xmlRelaxNGSetParserErrors(ctx->parser,
                                       (xmlRelaxNGValidityErrorFunc) fprintf,
                                       (xmlRelaxNGValidityWarningFunc) fprintf,
                                       stderr);
         }
 
         ctx->rng = xmlRelaxNGParse(ctx->parser);
         CRM_CHECK(ctx->rng != NULL,
                   crm_err("Could not find/parse %s", relaxng_file);
                   goto cleanup);
 
         ctx->valid = xmlRelaxNGNewValidCtxt(ctx->rng);
         CRM_CHECK(ctx->valid != NULL, goto cleanup);
 
         if (error_handler) {
             xmlRelaxNGSetValidErrors(ctx->valid,
                                      (xmlRelaxNGValidityErrorFunc) error_handler,
                                      (xmlRelaxNGValidityWarningFunc) error_handler,
                                      error_handler_context);
         } else {
             xmlRelaxNGSetValidErrors(ctx->valid,
                                      (xmlRelaxNGValidityErrorFunc) fprintf,
                                      (xmlRelaxNGValidityWarningFunc) fprintf,
                                      stderr);
         }
     }
 
     rc = xmlRelaxNGValidateDoc(ctx->valid, doc);
     if (rc > 0) {
         valid = FALSE;
 
     } else if (rc < 0) {
         crm_err("Internal libxml error during validation");
     }
 
   cleanup:
 
     if (cached_ctx) {
         *cached_ctx = ctx;
 
     } else {
         if (ctx->parser != NULL) {
             xmlRelaxNGFreeParserCtxt(ctx->parser);
         }
         if (ctx->valid != NULL) {
             xmlRelaxNGFreeValidCtxt(ctx->valid);
         }
         if (ctx->rng != NULL) {
             xmlRelaxNGFree(ctx->rng);
         }
         free(ctx);
     }
 
     return valid;
 }
 
 static void
 free_schema(gpointer data)
 {
     pcmk__schema_t *schema = data;
     relaxng_ctx_cache_t *ctx = NULL;
 
     switch (schema->validator) {
         case pcmk__schema_validator_none: // not cached
             break;
 
         case pcmk__schema_validator_rng: // cached
             ctx = (relaxng_ctx_cache_t *) schema->cache;
             if (ctx == NULL) {
                 break;
             }
 
             if (ctx->parser != NULL) {
                 xmlRelaxNGFreeParserCtxt(ctx->parser);
             }
 
             if (ctx->valid != NULL) {
                 xmlRelaxNGFreeValidCtxt(ctx->valid);
             }
 
             if (ctx->rng != NULL) {
                 xmlRelaxNGFree(ctx->rng);
             }
 
             free(ctx);
             schema->cache = NULL;
             break;
     }
 
     free(schema->name);
     free(schema->transform);
     free(schema->transform_enter);
 }
 
 /*!
  * \internal
  * \brief Clean up global memory associated with XML schemas
  */
 void
 crm_schema_cleanup(void)
 {
     g_list_free_full(known_schemas, free_schema);
     known_schemas = NULL;
 
     wrap_libxslt(true);
 }
 
+/*!
+ * \internal
+ * \brief Get schema list entry corresponding to a schema name
+ *
+ * \param[in] name  Name of schema to get
+ *
+ * \return Schema list entry corresponding to \p name, or NULL if unknown
+ */
+GList *
+pcmk__get_schema(const char *name)
+{
+    if (name == NULL) {
+        name = PCMK_VALUE_NONE;
+    }
+    for (GList *iter = known_schemas; iter != NULL; iter = iter->next) {
+        pcmk__schema_t *schema = iter->data;
+
+        if (pcmk__str_eq(name, schema->name, pcmk__str_casei)) {
+            return iter;
+        }
+    }
+    return NULL;
+}
+
 static gboolean
 validate_with(xmlNode *xml, pcmk__schema_t *schema, xmlRelaxNGValidityErrorFunc error_handler, void* error_handler_context)
 {
     gboolean valid = FALSE;
     char *file = NULL;
     relaxng_ctx_cache_t **cache = NULL;
 
     if (schema == NULL) {
         return FALSE;
     }
 
     if (schema->validator == pcmk__schema_validator_none) {
         return TRUE;
     }
 
     if (pcmk__str_eq(schema->name, "pacemaker-next", pcmk__str_none)) {
         crm_warn("The pacemaker-next schema is deprecated and will be removed "
                  "in a future release.");
     }
 
     file = pcmk__xml_artefact_path(pcmk__xml_artefact_ns_legacy_rng,
                                    schema->name);
 
     crm_trace("Validating with %s (type=%d)",
               pcmk__s(file, "missing schema"), schema->validator);
     switch (schema->validator) {
         case pcmk__schema_validator_rng:
             cache = (relaxng_ctx_cache_t **) &(schema->cache);
             valid = validate_with_relaxng(xml->doc, error_handler, error_handler_context, file, cache);
             break;
         default:
             crm_err("Unknown validator type: %d", schema->validator);
             break;
     }
 
     free(file);
     return valid;
 }
 
 static bool
 validate_with_silent(xmlNode *xml, pcmk__schema_t *schema)
 {
     bool rc, sl_backup = silent_logging;
     silent_logging = TRUE;
     rc = validate_with(xml, schema, (xmlRelaxNGValidityErrorFunc) xml_log, GUINT_TO_POINTER(LOG_ERR));
     silent_logging = sl_backup;
     return rc;
 }
 
 static void
 dump_file(const char *filename)
 {
 
     FILE *fp = NULL;
     int ch, line = 0;
 
     CRM_CHECK(filename != NULL, return);
 
     fp = fopen(filename, "r");
     if (fp == NULL) {
         crm_perror(LOG_ERR, "Could not open %s for reading", filename);
         return;
     }
 
     fprintf(stderr, "%4d ", ++line);
     do {
         ch = getc(fp);
         if (ch == EOF) {
             putc('\n', stderr);
             break;
         } else if (ch == '\n') {
             fprintf(stderr, "\n%4d ", ++line);
         } else {
             putc(ch, stderr);
         }
     } while (1);
 
     fclose(fp);
 }
 
 gboolean
 validate_xml_verbose(const xmlNode *xml_blob)
 {
     int fd = 0;
     xmlDoc *doc = NULL;
     xmlNode *xml = NULL;
     gboolean rc = FALSE;
     char *filename = NULL;
 
     filename = crm_strdup_printf("%s/cib-invalid.XXXXXX", pcmk__get_tmpdir());
 
     umask(S_IWGRP | S_IWOTH | S_IROTH);
     fd = mkstemp(filename);
     pcmk__xml_write_fd(xml_blob, filename, fd, false, NULL);
 
     dump_file(filename);
 
     doc = xmlReadFile(filename, NULL, 0);
     xml = xmlDocGetRootElement(doc);
     rc = validate_xml(xml, NULL, FALSE);
     free_xml(xml);
 
     unlink(filename);
     free(filename);
 
     return rc;
 }
 
 gboolean
 validate_xml(xmlNode *xml_blob, const char *validation, gboolean to_logs)
 {
     return pcmk__validate_xml(xml_blob, validation, to_logs ? (xmlRelaxNGValidityErrorFunc) xml_log : NULL, GUINT_TO_POINTER(LOG_ERR));
 }
 
 gboolean
 pcmk__validate_xml(xmlNode *xml_blob, const char *validation, xmlRelaxNGValidityErrorFunc error_handler, void* error_handler_context)
 {
     int version = 0;
 
     CRM_CHECK((xml_blob != NULL) && (xml_blob->doc != NULL), return FALSE);
 
     if (validation == NULL) {
         validation = crm_element_value(xml_blob, PCMK_XA_VALIDATE_WITH);
     }
 
     if (validation == NULL) {
         bool valid = FALSE;
 
         for (GList *iter = known_schemas; iter != NULL; iter = iter->next) {
             pcmk__schema_t *schema = iter->data;
 
             if (validate_with(xml_blob, schema, NULL, NULL)) {
                 valid = TRUE;
                 crm_xml_add(xml_blob, PCMK_XA_VALIDATE_WITH, schema->name);
                 crm_info("XML validated against %s", schema->name);
             }
         }
 
         return valid;
     }
 
     version = get_schema_version(validation);
     if (strcmp(validation, PCMK_VALUE_NONE) == 0) {
         return TRUE;
     } else if (version < g_list_length(known_schemas)) {
         pcmk__schema_t *schema = g_list_nth_data(known_schemas, version);
         return validate_with(xml_blob, schema, error_handler,
                              error_handler_context);
     }
 
     crm_err("Unknown validator: %s", validation);
     return FALSE;
 }
 
 /* With this arrangement, an attempt to identify the message severity
    as explicitly signalled directly from XSLT is performed in rather
    a smart way (no reliance on formatting string + arguments being
    always specified as ["%s", purposeful_string], as it can also be
    ["%s: %s", some_prefix, purposeful_string] etc. so every argument
    pertaining %s specifier is investigated), and if such a mark found,
    the respective level is determined and, when the messages are to go
    to the native logs, the mark itself gets dropped
    (by the means of string shift).
 
    NOTE: whether the native logging is the right sink is decided per
          the ctx parameter -- NULL denotes this case, otherwise it
          carries a pointer to the numeric expression of the desired
          target logging level (messages with higher level will be
          suppressed)
 
    NOTE: on some architectures, this string shift may not have any
          effect, but that's an acceptable tradeoff
 
    The logging level for not explicitly designated messages
    (suspicious, likely internal errors or some runaways) is
    LOG_WARNING.
  */
 static void G_GNUC_PRINTF(2, 3)
 cib_upgrade_err(void *ctx, const char *fmt, ...)
 {
     va_list ap, aq;
     char *arg_cur;
 
     bool found = FALSE;
     const char *fmt_iter = fmt;
     uint8_t msg_log_level = LOG_WARNING;  /* default for runaway messages */
     const unsigned * log_level = (const unsigned *) ctx;
     enum {
         escan_seennothing,
         escan_seenpercent,
     } scan_state = escan_seennothing;
 
     va_start(ap, fmt);
     va_copy(aq, ap);
 
     while (!found && *fmt_iter != '\0') {
         /* while casing schema borrowed from libqb:qb_vsnprintf_serialize */
         switch (*fmt_iter++) {
         case '%':
             if (scan_state == escan_seennothing) {
                 scan_state = escan_seenpercent;
             } else if (scan_state == escan_seenpercent) {
                 scan_state = escan_seennothing;
             }
             break;
         case 's':
             if (scan_state == escan_seenpercent) {
                 scan_state = escan_seennothing;
                 arg_cur = va_arg(aq, char *);
                 if (arg_cur != NULL) {
                     switch (arg_cur[0]) {
                     case 'W':
                         if (!strncmp(arg_cur, "WARNING: ",
                                      sizeof("WARNING: ") - 1)) {
                             msg_log_level = LOG_WARNING;
                         }
                         if (ctx == NULL) {
                             memmove(arg_cur, arg_cur + sizeof("WARNING: ") - 1,
                                     strlen(arg_cur + sizeof("WARNING: ") - 1) + 1);
                         }
                         found = TRUE;
                         break;
                     case 'I':
                         if (!strncmp(arg_cur, "INFO: ",
                                      sizeof("INFO: ") - 1)) {
                             msg_log_level = LOG_INFO;
                         }
                         if (ctx == NULL) {
                             memmove(arg_cur, arg_cur + sizeof("INFO: ") - 1,
                                     strlen(arg_cur + sizeof("INFO: ") - 1) + 1);
                         }
                         found = TRUE;
                         break;
                     case 'D':
                         if (!strncmp(arg_cur, "DEBUG: ",
                                      sizeof("DEBUG: ") - 1)) {
                             msg_log_level = LOG_DEBUG;
                         }
                         if (ctx == NULL) {
                             memmove(arg_cur, arg_cur + sizeof("DEBUG: ") - 1,
                                     strlen(arg_cur + sizeof("DEBUG: ") - 1) + 1);
                         }
                         found = TRUE;
                         break;
                     }
                 }
             }
             break;
         case '#': case '-': case ' ': case '+': case '\'': case 'I': case '.':
         case '0': case '1': case '2': case '3': case '4':
         case '5': case '6': case '7': case '8': case '9':
         case '*':
             break;
         case 'l':
         case 'z':
         case 't':
         case 'j':
         case 'd': case 'i':
         case 'o':
         case 'u':
         case 'x': case 'X':
         case 'e': case 'E':
         case 'f': case 'F':
         case 'g': case 'G':
         case 'a': case 'A':
         case 'c':
         case 'p':
             if (scan_state == escan_seenpercent) {
                 (void) va_arg(aq, void *);  /* skip forward */
                 scan_state = escan_seennothing;
             }
             break;
         default:
             scan_state = escan_seennothing;
             break;
         }
     }
 
     if (log_level != NULL) {
         /* intention of the following offset is:
            cibadmin -V -> start showing INFO labelled messages */
         if (*log_level + 4 >= msg_log_level) {
             vfprintf(stderr, fmt, ap);
         }
     } else {
         PCMK__XML_LOG_BASE(msg_log_level, TRUE, 0, "CIB upgrade: ", fmt, ap);
     }
 
     va_end(aq);
     va_end(ap);
 }
 
 /*!
  * \internal
  * \brief Apply a single XSL transformation to given XML
  *
  * \param[in] xml        XML to transform
  * \param[in] transform  XSL name
  * \param[in] to_logs    If false, certain validation errors will be sent to
  *                       stderr rather than logged
  *
  * \return Transformed XML on success, otherwise NULL
  */
 static xmlNode *
 apply_transformation(const xmlNode *xml, const char *transform,
                      gboolean to_logs)
 {
     char *xform = NULL;
     xmlNode *out = NULL;
     xmlDocPtr res = NULL;
     xsltStylesheet *xslt = NULL;
 
     xform = pcmk__xml_artefact_path(pcmk__xml_artefact_ns_legacy_xslt,
                                     transform);
 
     /* for capturing, e.g., what's emitted via <xsl:message> */
     if (to_logs) {
         xsltSetGenericErrorFunc(NULL, cib_upgrade_err);
     } else {
         xsltSetGenericErrorFunc(&crm_log_level, cib_upgrade_err);
     }
 
     xslt = xsltParseStylesheetFile((pcmkXmlStr) xform);
     CRM_CHECK(xslt != NULL, goto cleanup);
 
     res = xsltApplyStylesheet(xslt, xml->doc, NULL);
     CRM_CHECK(res != NULL, goto cleanup);
 
     xsltSetGenericErrorFunc(NULL, NULL);  /* restore default one */
 
     out = xmlDocGetRootElement(res);
 
   cleanup:
     if (xslt) {
         xsltFreeStylesheet(xslt);
     }
 
     free(xform);
 
     return out;
 }
 
 /*!
  * \internal
  * \brief Perform all transformations needed to upgrade XML to next schema
  *
  * A schema upgrade can require up to three XSL transformations: an "enter"
  * transform, the main upgrade transform, and a "leave" transform. Perform
  * all needed transforms to upgrade given XML to the next schema.
  *
  * \param[in] original_xml  XML to transform
  * \param[in] schema_index  Index of schema that successfully validates
  *                          \p original_xml
  * \param[in] to_logs       If false, certain validation errors will be sent to
  *                          stderr rather than logged
  *
  * \return XML result of schema transforms if successful, otherwise NULL
  */
 static xmlNode *
 apply_upgrade(const xmlNode *original_xml, int schema_index, gboolean to_logs)
 {
     pcmk__schema_t *schema = g_list_nth_data(known_schemas, schema_index);
     pcmk__schema_t *upgraded_schema = g_list_nth_data(known_schemas,
                                                       schema_index + 1);
     bool transform_onleave = false;
     char *transform_leave;
     const xmlNode *xml = original_xml;
     xmlNode *upgrade = NULL;
     xmlNode *final = NULL;
     xmlRelaxNGValidityErrorFunc error_handler = NULL;
 
     CRM_ASSERT((schema != NULL) && (upgraded_schema != NULL));
 
     if (to_logs) {
         error_handler = (xmlRelaxNGValidityErrorFunc) xml_log;
     }
 
     transform_onleave = schema->transform_onleave;
     if (schema->transform_enter != NULL) {
         crm_debug("Upgrading schema from %s to %s: "
                   "applying pre-upgrade XSL transform %s",
                   schema->name, upgraded_schema->name, schema->transform_enter);
         upgrade = apply_transformation(xml, schema->transform_enter, to_logs);
         if (upgrade == NULL) {
             crm_warn("Pre-upgrade XSL transform %s failed, "
                      "will skip post-upgrade transform",
                      schema->transform_enter);
             transform_onleave = FALSE;
         } else {
             xml = upgrade;
         }
     }
 
 
     crm_debug("Upgrading schema from %s to %s: "
               "applying upgrade XSL transform %s",
               schema->name, upgraded_schema->name, schema->transform);
     final = apply_transformation(xml, schema->transform, to_logs);
     if (upgrade != xml) {
         free_xml(upgrade);
         upgrade = NULL;
     }
 
     if ((final != NULL) && transform_onleave) {
         upgrade = final;
         /* following condition ensured in add_schema_by_version */
         CRM_ASSERT(schema->transform_enter != NULL);
         transform_leave = strdup(schema->transform_enter);
         /* enter -> leave */
         memcpy(strrchr(transform_leave, '-') + 1, "leave", sizeof("leave") - 1);
         crm_debug("Upgrading schema from %s to %s: "
                   "applying post-upgrade XSL transform %s",
                   schema->name, upgraded_schema->name, transform_leave);
         final = apply_transformation(upgrade, transform_leave, to_logs);
         if (final == NULL) {
             crm_warn("Ignoring failure of post-upgrade XSL transform %s",
                      transform_leave);
             final = upgrade;
         } else {
             free_xml(upgrade);
         }
         free(transform_leave);
     }
 
     if (final == NULL) {
         return NULL;
     }
 
     // Ensure result validates with its new schema
     if (!validate_with(final, upgraded_schema, error_handler,
                        GUINT_TO_POINTER(LOG_ERR))) {
         crm_err("Schema upgrade from %s to %s failed: "
                 "XSL transform %s produced an invalid configuration",
                 schema->name, upgraded_schema->name, schema->transform);
         crm_log_xml_debug(final, "bad-transform-result");
         free_xml(final);
         return NULL;
     }
 
     crm_info("Schema upgrade from %s to %s succeeded",
              schema->name, upgraded_schema->name);
     return final;
 }
 
 const char *
 get_schema_name(int version)
 {
     pcmk__schema_t *schema = g_list_nth_data(known_schemas, version);
 
     return (schema != NULL)? schema->name : "unknown";
 }
 
 int
 get_schema_version(const char *name)
 {
     int lpc = 0;
 
     if (name == NULL) {
         name = PCMK_VALUE_NONE;
     }
 
     for (GList *iter = known_schemas; iter != NULL; iter = iter->next) {
         pcmk__schema_t *schema = iter->data;
 
         if (pcmk__str_eq(name, schema->name, pcmk__str_casei)) {
             return lpc;
         }
 
         lpc++;
     }
 
     return -1;
 }
 
 /* set which validation to use */
 /*!
  * \brief Update CIB XML to latest schema that validates it
  *
  * \param[in,out] xml_blob   XML to update (may be freed and replaced after
  *                           being transformed)
  * \param[out]    best       If not NULL, set to schema index of latest schema
  *                           that validates \p xml_blob
  * \param[in]     max        If positive, do not update \p xml_blob to any
  *                           schema past this index
  * \param[in]     transform  If false, do not update \p xml_blob to any schema
  *                           that requires an XSL transform
  * \param[in]     to_logs    If false, certain validation errors will be sent to
  *                           stderr rather than logged
  *
  * \return Legacy Pacemaker return code
  */
 int
 update_validation(xmlNode **xml_blob, int *best, int max, gboolean transform,
                   gboolean to_logs)
 {
     xmlNode *xml = NULL;
     char *value = NULL;
     int max_stable_schemas = xml_latest_schema_index();
     int lpc = 0, match = -1, rc = pcmk_ok;
     int local_best = 0;
     int next = -1;  /* -1 denotes "inactive" value */
     xmlRelaxNGValidityErrorFunc error_handler = 
         to_logs ? (xmlRelaxNGValidityErrorFunc) xml_log : NULL;
 
     if (best != NULL) {
         *best = 0;
     }
 
     CRM_CHECK((xml_blob != NULL) && (*xml_blob != NULL)
               && ((*xml_blob)->doc != NULL),
               return -EINVAL);
 
     xml = *xml_blob;
     value = crm_element_value_copy(xml, PCMK_XA_VALIDATE_WITH);
 
     if (value != NULL) {
         match = get_schema_version(value);
         if (match >= max_stable_schemas) {
             // No higher version is available
             free(value);
             if (best != NULL) {
                 *best = match;
             }
             return pcmk_ok;
         }
 
         lpc = match;
         if (lpc >= 0 && transform == FALSE) {
             local_best = lpc++;
 
         } else if (lpc < 0) {
             crm_debug("Unknown validation schema");
             lpc = 0;
         }
     }
 
     while (lpc <= max_stable_schemas) {
         /* FIXME: This will cause us to walk the known_schemas list every time
          * this loop iterates, which is not ideal.  However, for now it's a lot
          * easier than trying to get all the loop indices we're using here
          * sorted out and working correctly.
          */
         pcmk__schema_t *schema = g_list_nth_data(known_schemas, lpc);
         pcmk__schema_t *next_schema = NULL;
         xmlNode *upgrade = NULL;
 
         crm_debug("Testing '%s' validation (%d of %d)",
                   pcmk__s(schema->name, "<unset>"), lpc, max_stable_schemas);
 
         if (validate_with(xml, schema, error_handler, GUINT_TO_POINTER(LOG_ERR)) == FALSE) {
             if (next != -1) {
                 crm_info("Configuration not valid for schema: %s",
                          schema->name);
                 next = -1;
             } else {
                 crm_trace("%s validation failed", pcmk__s(schema->name, "<unset>"));
             }
             if (local_best > 0) {
                 /* we've satisfied the validation, no need to check further */
                 break;
             }
             rc = -pcmk_err_schema_validation;
             lpc++; // Try again with the next higher schema
             continue;
         }
 
         if (next != -1) {
             crm_debug("Configuration valid for schema: %s", schema->name);
             next = -1;
         }
         rc = pcmk_ok;
         local_best = lpc;
 
         if (!transform) {
             /* Validation with this schema succeeded. We are not doing
              * transforms, so try the next schema using the same XML.
              */
             lpc++;
             continue;
         }
 
         next = lpc+1;
 
         if (next > max_stable_schemas) {
             /* There is no next version */
             crm_trace("Stopping at %s", schema->name);
             break;
         }
 
         if (max > 0 && (lpc == max || next > max)) {
             crm_trace("Upgrade limit reached at %s (lpc=%d, next=%d, max=%d)",
                       schema->name, lpc, next, max);
             break;
         }
 
         next_schema = g_list_nth_data(known_schemas, next);
         CRM_ASSERT(next_schema != NULL);
 
         if ((schema->transform == NULL)
             || validate_with_silent(xml, next_schema)) {
             /* The next schema either doesn't require a transform, or validates
              * successfully even without doing the transform. We can skip the
              * transform and use it with the same XML in the next iteration.
              */
             crm_debug("%s-style configuration is also valid for %s",
                        schema->name, next_schema->name);
             lpc = next;
             continue;
         }
 
         upgrade = apply_upgrade(xml, lpc, to_logs);
         if (upgrade == NULL) {
             rc = -pcmk_err_transform_failed;
         } else {
             lpc = next;
             local_best = next;
             free_xml(xml);
             xml = upgrade;
         }
         next = -1;
         if (rc != pcmk_ok) {
             /* The transform failed, so this schema can't be used. Later
              * schemas are unlikely to validate, but try anyway until we
              * run out of options.
              */
             lpc++;
         }
     }
 
     if ((local_best > 0) && (local_best > match)) {
         pcmk__schema_t *best_schema = g_list_nth_data(known_schemas,
                                                       local_best);
 
         crm_info("%s the configuration from %s to %s",
                    transform?"Transformed":"Upgraded", pcmk__s(value, "<none>"),
                    best_schema->name);
         crm_xml_add(xml, PCMK_XA_VALIDATE_WITH, best_schema->name);
     }
 
     *xml_blob = xml;
     free(value);
 
     if (best != NULL) {
         *best = local_best;
     }
     return rc;
 }
 
 gboolean
 cli_config_update(xmlNode **xml, int *best_version, gboolean to_logs)
 {
     gboolean rc = TRUE;
     const char *value = crm_element_value(*xml, PCMK_XA_VALIDATE_WITH);
     char *const orig_value = strdup(value == NULL ? "(none)" : value);
 
     int version = get_schema_version(value);
     int orig_version = version;
     int min_version = pcmk__find_x_0_schema_index();
 
     if (version < min_version) {
         // Current configuration schema is not acceptable, try to update
         xmlNode *converted = NULL;
 
         converted = pcmk__xml_copy(NULL, *xml);
         update_validation(&converted, &version, 0, TRUE, to_logs);
 
         value = crm_element_value(converted, PCMK_XA_VALIDATE_WITH);
         if (version < min_version) {
             // Updated configuration schema is still not acceptable
 
             if (version < orig_version || orig_version == -1) {
                 // We couldn't validate any schema at all
                 if (to_logs) {
                     pcmk__config_err("Cannot upgrade configuration (claiming "
                                      "schema %s) to at least %s because it "
                                      "does not validate with any schema from "
                                      "%s to %s",
                                      orig_value,
                                      get_schema_name(min_version),
                                      get_schema_name(orig_version),
                                      xml_latest_schema());
                 } else {
                     fprintf(stderr, "Cannot upgrade configuration (claiming "
                                     "schema %s) to at least %s because it "
                                     "does not validate with any schema from "
                                     "%s to %s\n",
                                     orig_value,
                                     get_schema_name(min_version),
                                     get_schema_name(orig_version),
                                     xml_latest_schema());
                 }
             } else {
                 // We updated configuration successfully, but still too low
                 if (to_logs) {
                     pcmk__config_err("Cannot upgrade configuration (claiming "
                                      "schema %s) to at least %s because it "
                                      "would not upgrade past %s",
                                      orig_value,
                                      get_schema_name(min_version),
                                      pcmk__s(value, "unspecified version"));
                 } else {
                     fprintf(stderr, "Cannot upgrade configuration (claiming "
                                     "schema %s) to at least %s because it "
                                     "would not upgrade past %s\n",
                                     orig_value,
                                     get_schema_name(min_version),
                                     pcmk__s(value, "unspecified version"));
                 }
             }
 
             free_xml(converted);
             converted = NULL;
             rc = FALSE;
 
         } else {
             // Updated configuration schema is acceptable
             free_xml(*xml);
             *xml = converted;
 
             if (version < xml_latest_schema_index()) {
                 if (to_logs) {
                     pcmk__config_warn("Configuration with schema %s was "
                                       "internally upgraded to acceptable (but "
                                       "not most recent) %s",
                                       orig_value, get_schema_name(version));
                 }
             } else {
                 if (to_logs) {
                     crm_info("Configuration with schema %s was internally "
                              "upgraded to latest version %s",
                              orig_value, get_schema_name(version));
                 }
             }
         }
 
     } else if (version >= get_schema_version(PCMK_VALUE_NONE)) {
         // Schema validation is disabled
         if (to_logs) {
             pcmk__config_warn("Schema validation of configuration is disabled "
                               "(enabling is encouraged and prevents common "
                               "misconfigurations)");
 
         } else {
             fprintf(stderr, "Schema validation of configuration is disabled "
                             "(enabling is encouraged and prevents common "
                             "misconfigurations)\n");
         }
     }
 
     if (best_version) {
         *best_version = version;
     }
 
     free(orig_value);
     return rc;
 }
 
 /*!
  * \internal
  * \brief Return a list of all schema files and any associated XSLT files
  *        later than the given one
  * \brief Return a list of all schema versions later than the given one
  *
  * \param[in] schema The schema to compare against (for example,
  *                   "pacemaker-3.1.rng" or "pacemaker-3.1")
  *
  * \note The caller is responsible for freeing both the returned list and
  *       the elements of the list
  */
 GList *
 pcmk__schema_files_later_than(const char *name)
 {
     GList *lst = NULL;
     pcmk__schema_version_t ver;
 
     if (!version_from_filename(name, &ver)) {
         return lst;
     }
 
     for (GList *iter = g_list_nth(known_schemas, xml_latest_schema_index());
          iter != NULL; iter = iter->prev) {
         pcmk__schema_t *schema = iter->data;
         char *s = NULL;
 
         if (schema_cmp(ver, schema->version) != -1) {
             continue;
         }
 
         s = crm_strdup_printf("%s.rng", schema->name);
         lst = g_list_prepend(lst, s);
 
         if (schema->transform != NULL) {
             char *xform = crm_strdup_printf("%s.xsl", schema->transform);
             lst = g_list_prepend(lst, xform);
         }
 
         if (schema->transform_enter != NULL) {
             char *enter = crm_strdup_printf("%s.xsl", schema->transform_enter);
 
             lst = g_list_prepend(lst, enter);
 
             if (schema->transform_onleave) {
                 int last_dash = strrchr(enter, '-') - enter;
                 char *leave = crm_strdup_printf("%.*s-leave.xsl", last_dash, enter);
 
                 lst = g_list_prepend(lst, leave);
             }
         }
     }
 
     return lst;
 }
 
 static void
 append_href(xmlNode *xml, void *user_data)
 {
     GList **list = user_data;
     char *href = crm_element_value_copy(xml, "href");
 
     if (href == NULL) {
         return;
     }
     *list = g_list_prepend(*list, href);
 }
 
 static void
 external_refs_in_schema(GList **list, const char *contents)
 {
     /* local-name()= is needed to ignore the xmlns= setting at the top of
      * the XML file.  Otherwise, the xpath query will always return nothing.
      */
     const char *search = "//*[local-name()='externalRef'] | //*[local-name()='include']";
     xmlNode *xml = pcmk__xml_parse(contents);
 
     crm_foreach_xpath_result(xml, search, append_href, list);
     free_xml(xml);
 }
 
 static int
 read_file_contents(const char *file, char **contents)
 {
     int rc = pcmk_rc_ok;
     char *path = NULL;
 
     if (pcmk__ends_with(file, ".rng")) {
         path = pcmk__xml_artefact_path(pcmk__xml_artefact_ns_legacy_rng, file);
     } else {
         path = pcmk__xml_artefact_path(pcmk__xml_artefact_ns_legacy_xslt, file);
     }
 
     rc = pcmk__file_contents(path, contents);
 
     free(path);
     return rc;
 }
 
 static void
 add_schema_file_to_xml(xmlNode *parent, const char *file, GList **already_included)
 {
     char *contents = NULL;
     char *path = NULL;
     xmlNode *file_node = NULL;
     GList *includes = NULL;
     int rc = pcmk_rc_ok;
 
     /* If we already included this file, don't do so again. */
     if (g_list_find_custom(*already_included, file, (GCompareFunc) strcmp) != NULL) {
         return;
     }
 
     /* Ensure whatever file we were given has a suffix we know about.  If not,
      * just assume it's an RNG file.
      */
     if (!pcmk__ends_with(file, ".rng") && !pcmk__ends_with(file, ".xsl")) {
         path = crm_strdup_printf("%s.rng", file);
     } else {
         path = pcmk__str_copy(file);
     }
 
     rc = read_file_contents(path, &contents);
     if (rc != pcmk_rc_ok || contents == NULL) {
         crm_warn("Could not read schema file %s: %s", file, pcmk_rc_str(rc));
         free(path);
         return;
     }
 
     /* Create a new <file path="..."> node with the contents of the file
      * as a CDATA block underneath it.
      */
     file_node = pcmk__xe_create(parent, PCMK_XA_FILE);
     crm_xml_add(file_node, PCMK_XA_PATH, path);
     *already_included = g_list_prepend(*already_included, path);
 
     xmlAddChild(file_node, xmlNewCDataBlock(parent->doc, (pcmkXmlStr) contents,
                                             strlen(contents)));
 
     /* Scan the file for any <externalRef> or <include> nodes and build up
      * a list of the files they reference.
      */
     external_refs_in_schema(&includes, contents);
 
     /* For each referenced file, recurse to add it (and potentially anything it
      * references, ...) to the XML.
      */
     for (GList *iter = includes; iter != NULL; iter = iter->next) {
         add_schema_file_to_xml(parent, iter->data, already_included);
     }
 
     free(contents);
     g_list_free_full(includes, free);
 }
 
 /*!
  * \internal
  * \brief Add an XML schema file and all the files it references as children
  *        of a given XML node
  *
  * \param[in,out] parent            The parent XML node
  * \param[in] name                  The schema version to compare against
  *                                  (for example, "pacemaker-3.1" or "pacemaker-3.1.rng")
  * \param[in,out] already_included  A list of names that have already been added
  *                                  to the parent node.
  *
  * \note The caller is responsible for freeing both the returned list and
  *       the elements of the list
  */
 void
 pcmk__build_schema_xml_node(xmlNode *parent, const char *name, GList **already_included)
 {
     /* First, create an unattached node to add all the schema files to as children. */
     xmlNode *schema_node = pcmk__xe_create(NULL, PCMK__XA_SCHEMA);
 
     crm_xml_add(schema_node, PCMK_XA_VERSION, name);
     add_schema_file_to_xml(schema_node, name, already_included);
 
     /* Then, if we actually added any children, attach the node to parent.  If
      * we did not add any children (for instance, name was invalid), this prevents
      * us from returning a document with additional empty children.
      */
     if (schema_node->children != NULL) {
         xmlAddChild(parent, schema_node);
     } else {
         free_xml(schema_node);
     }
 }
 
 /*!
  * \internal
  * \brief Return the directory containing any extra schema files that a
  *        Pacemaker Remote node fetched from the cluster
  */
 const char *
 pcmk__remote_schema_dir(void)
 {
     const char *dir = pcmk__env_option(PCMK__ENV_REMOTE_SCHEMA_DIRECTORY);
 
     if (pcmk__str_empty(dir)) {
         return PCMK__REMOTE_SCHEMA_DIR;
     }
 
     return dir;
 }
diff --git a/lib/common/tests/schemas/Makefile.am b/lib/common/tests/schemas/Makefile.am
index d9206e589e..2654cfd8cb 100644
--- a/lib/common/tests/schemas/Makefile.am
+++ b/lib/common/tests/schemas/Makefile.am
@@ -1,88 +1,89 @@
 #
-# Copyright 2023 the Pacemaker project contributors
+# Copyright 2023-2024 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 $(top_srcdir)/mk/tap.mk
 include $(top_srcdir)/mk/unittest.mk
 
 CFLAGS += -DPCMK__TEST_SCHEMA_DIR='"$(abs_builddir)/schemas"'
 
 # Add "_test" to the end of all test program names to simplify .gitignore.
 
 # These tests share a schema subdirectory
 SHARED_SCHEMA_TESTS =	crm_schema_init_test			\
 			get_schema_name_test			\
 			get_schema_version_test			\
 			pcmk__build_schema_xml_node_test	\
+			pcmk__get_schema_test 			\
 			pcmk__schema_files_later_than_test
 
 # This test has its own schema directory
 FIND_X_0_SCHEMA_TEST =	pcmk__find_x_0_schema_index_test
 
 check_PROGRAMS =	$(SHARED_SCHEMA_TESTS) $(FIND_X_0_SCHEMA_TEST)
 
 TESTS = $(check_PROGRAMS)
 
 $(SHARED_SCHEMA_TESTS): setup-schema-dir
 
 $(FIND_X_0_SCHEMA_TEST): setup-find_x_0-schema-dir
 
 # Set up a temporary schemas/ directory containing only some of the full set of
 # pacemaker schema files.  This lets us know exactly how many schemas are present,
 # allowing us to write tests without having to make changes when new schemas are
 # added.
 #
 # This directory contains the following:
 #
 # * pacemaker-next.rng   - Used to verify that this sorts before all versions
 # * upgrade-*.xsl        - Required by various schema versions
 # * pacemaker-[0-9]*.rng - We're only pulling in 15 schemas, which is enough
 #                          to get everything through pacemaker-3.0.rng.  This
 #                          includes 2.10, needed so we can check that versions
 #                          are compared as numbers instead of strings.
 # * other RNG files      - This catches everything except the pacemaker-*rng
 #                          files.  These files are included by the top-level
 #                          pacemaker-*rng files, so we need them for tests.
 #                          This will glob more than we need, but the extra ones
 #                          won't get in the way.
 
 LINK_FILES =	$(abs_top_builddir)/xml/pacemaker-next.rng	\
 		$(abs_top_builddir)/xml/upgrade-*.xsl
 ROOT_RNGS = 	$(shell ls -1v $(abs_top_builddir)/xml/pacemaker-[0-9]*.rng | head -15)
 INCLUDED_RNGS = $(shell ls -1 $(top_srcdir)/xml/*.rng | grep -v pacemaker-[0-9])
 
 # Most tests share a common, read-only schema directory
 .PHONY: setup-schema-dir
 setup-schema-dir:
 	$(MKDIR_P) schemas
 	( cd schemas ; \
 	  ln -sf $(LINK_FILES) . ; \
 	  for f in $(ROOT_RNGS); do \
 		ln -sf $$f $$(basename $$f); \
 	  done ; \
 	  for f in $(INCLUDED_RNGS); do \
 		ln -sf ../$$f $$(basename $$f); \
 	  done )
 
 # pcmk__find_x_0_schema_index_test moves schema files around, so it needs its
 # own directory, otherwise other tests run in parallel could fail.
 .PHONY: setup-find_x_0-schema-dir
 setup-find_x_0-schema-dir:
 	$(MKDIR_P) schemas/find_x_0
 	( cd schemas/find_x_0 ; \
 	  ln -sf $(LINK_FILES) . ; \
 	  for f in $(ROOT_RNGS); do \
 		ln -sf $$f $$(basename $$f); \
 	  done ; \
 	  for f in $(INCLUDED_RNGS); do \
 		ln -sf ../$$f $$(basename $$f); \
 	  done )
 
 .PHONY: clean-local
 clean-local:
 	-rm -rf schemas
diff --git a/lib/common/tests/schemas/pcmk__get_schema_test.c b/lib/common/tests/schemas/pcmk__get_schema_test.c
new file mode 100644
index 0000000000..edc0c995e9
--- /dev/null
+++ b/lib/common/tests/schemas/pcmk__get_schema_test.c
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2023-2024 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 <crm/common/xml.h>
+#include <crm/common/unittest_internal.h>
+#include <crm/common/xml_internal.h>
+#include "crmcommon_private.h"
+
+static int
+setup(void **state)
+{
+    setenv("PCMK_schema_directory", PCMK__TEST_SCHEMA_DIR, 1);
+    crm_schema_init();
+    return 0;
+}
+
+static int
+teardown(void **state)
+{
+    crm_schema_cleanup();
+    unsetenv("PCMK_schema_directory");
+    return 0;
+}
+
+static void
+assert_schema(const char *name, int expected_index)
+{
+    GList *schema_entry = NULL;
+    pcmk__schema_t *schema = NULL;
+
+    schema_entry = pcmk__get_schema(name);
+    assert_non_null(schema_entry);
+
+    schema = schema_entry->data;
+    assert_non_null(schema);
+
+    assert_int_equal(schema->schema_index, expected_index);
+}
+
+static void
+unknown_schema(void **state)
+{
+    assert_null(pcmk__get_schema(""));
+    assert_null(pcmk__get_schema("blahblah"));
+    assert_null(pcmk__get_schema("pacemaker-2.47"));
+    assert_null(pcmk__get_schema("pacemaker-47.0"));
+}
+
+static void
+known_schema(void **state)
+{
+    assert_schema(NULL, 16); // defaults to "none"
+    assert_schema("pacemaker-1.0", 0);
+    assert_schema("PACEMAKER-1.0", 0);
+    assert_schema("pacemaker-1.2", 1);
+    assert_schema("pacemaker-2.0", 3);
+    assert_schema("pAcEmAkEr-2.0", 3);
+    assert_schema("pacemaker-2.5", 8);
+    assert_schema("pacemaker-3.0", 14);
+    assert_schema("paceMAKER-3.0", 14);
+}
+
+PCMK__UNIT_TEST(setup, teardown,
+                cmocka_unit_test(unknown_schema),
+                cmocka_unit_test(known_schema));