fix(scheduler): a re-applied config updates queues in place and frees the config it replaces (bug-1886)

sched_init_queues() built a new xce_sched_queue_t for every configured
queue on each config apply; for an existing name the add failed, and the
unregistered queue was handed to the picker (which then tracked it
instead of the registered one) and leaked, while the registered queue
kept its old settings. It now updates the registered queue in place, as
QUEUE_UPDATE's core_upsert_queue() does.

sched_init_config() and the core's CONFIG_PUSH replaced core->config
without freeing the previous one. The core owns its config, and nothing
points into a replaced one (queues, resources, groups and limits built
from it are copies), so it is freed once the new one is applied.

Test: test_config_reapply_updates_queue_in_place_and_frees_old RED before
(priority stayed 1), GREEN after.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

src/scheduler/src/core_events.c

@@ -834,6 +834,7 @@ int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events,
         switch (ev->type) {
         case XCE_CORE_EVENT_CONFIG_PUSH: {
             xce_conf4sched_t* config = ev->u.config_push.config;
+            xce_conf4sched_t* older  = core->config;
 
             core->config = config;
             core->configver++;
@@ -847,6 +848,12 @@ int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events,
             if ((picker != NULL) && (picker->picker_iface->on_config != NULL)) {
                 picker->picker_iface->on_config(picker, config);
             }
+
+            /* bug-1886: the core owns its config; the replaced one is no
+             * longer referenced (everything built from it is a copy) */
+            if ((older != NULL) && (older != config)) {
+                free_Conf4Sched_ptr(&older);
+            }
             break;
         }
 

src/scheduler/src/scheduler.c

@@ -1042,10 +1042,34 @@ static int sched_init_queues(xce_scheduler_t* sched)
     }
 
     for (i = 0; i < config->n_queues; i++) {
-        xce_sched_queue_t* q = make_sched_queue(sched, config->queues[i]);
+        xce_queue_t*       cq = config->queues[i];
+        xce_sched_queue_t* q  = NULL;
 
-        plugin_queue_update(sched, q);
+        if ((cq == NULL) || (cq->queue == NULL)) {
-        sched_add_queue(&(sched->core), q);
+            continue;
+        }
+
+        /* bug-1886: a re-applied config updates the registered queue in
+         * place (as QUEUE_UPDATE does, core_upsert_queue()). Building a
+         * new one made the add fail on the existing name, and the new,
+         * unregistered queue was handed to the picker and leaked while the
+         * registered one kept its old settings. */
+        q = sched_get_queue(&(sched->core), cq->queue);
+        if (q != NULL) {
+            q->attrs           = cq->attrs;
+            q->status          = cq->status;
+            q->priority        = cq->priority;
+            q->accept_interval = cq->accept_interval;
+            q->sched_delay     = cq->sched_delay;
+            plugin_queue_update(sched, q);
+        } else {
+            q = make_sched_queue(sched, cq);
+            if (q == NULL) {
+                continue;
+            }
+            plugin_queue_update(sched, q);
+            sched_add_queue(&(sched->core), q);
+        }
 
         if (picker != NULL) {
             picker->picker_iface->on_queue_update(picker, q);
@@ -1056,6 +1080,8 @@ static int sched_init_queues(xce_scheduler_t* sched)
 }
 int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
 {
+    xce_conf4sched_t* older;
+
     /* bug-1784: register the cluster's NAMED usergroups/hostgroups (from
      * users.yaml's "usergroups:"/hosts.yaml's "hostgroups:") into
      * core.ugmgr/core.hgmgr BEFORE plugin_config_update() below -- a limit
@@ -1067,6 +1093,7 @@ int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
     groupmgr_load_configs(sched->core.hgmgr, config->n_hgroups, config->hgroups);
 
     plugin_config_update(sched, config);
+    older              = sched->core.config;
     sched->core.config = config;
 
     sched_init_resources(sched);
@@ -1075,6 +1102,13 @@ int sched_init_config(xce_scheduler_t* sched, xce_conf4sched_t* config)
 
     sched->core.configver++; /* increase the version of configuration */
 
+    /* bug-1886: the core owns its config (freed in xce_sched_core_fini());
+     * the one this replaces is no longer referenced -- the queues,
+     * resources, groups and limits built from it are copies. */
+    if ((older != NULL) && (older != config)) {
+        free_Conf4Sched_ptr(&older);
+    }
+
     return 0;
 }
 

src/scheduler/testsuite/test_config_push_deferred.c

@@ -31,6 +31,10 @@
 
 #include <unity/unity.h>
 
+#if defined(__SANITIZE_ADDRESS__)
+#include <sanitizer/asan_interface.h>
+#endif
+
 #include "controller.capnp.h"
 #include "plugin.h"
 #include "resobj.h"
@@ -249,10 +253,83 @@ static void test_applied_config_push_is_recorded(void)
     printf("test_applied_config_push_is_recorded: PASS\n");
 }
 
+/* bug-1886: stashes and applies a config carrying one queue, QUEUE_NAME,
+ * at `priority`; returns the applied config. */
+static xce_conf4sched_t* apply_queue_config(xce_scheduler_t* sched,
+                                            int32_t          priority)
+{
+    xce_queue_t*      q      = (xce_queue_t*)calloc(1, sizeof(*q));
+    xce_queue_t**     queues = (xce_queue_t**)calloc(1, sizeof(*queues));
+    xce_conf4sched_t* config = (xce_conf4sched_t*)calloc(1, sizeof(*config));
+
+    TEST_ASSERT_NOT_NULL(q);
+    TEST_ASSERT_NOT_NULL(queues);
+    TEST_ASSERT_NOT_NULL(config);
+
+    q->queue         = strdup(QUEUE_NAME);
+    q->priority      = priority;
+    queues[0]        = q;
+    config->n_queues = 1;
+    config->queues   = queues;
+
+    TEST_ASSERT_EQUAL_INT(0, sched_stash_config(sched, config));
+    TEST_ASSERT_EQUAL_INT(1, sched_apply_pending_config(sched));
+    TEST_ASSERT_EQUAL_PTR(config, sched->core.config);
+
+    return config;
+}
+
+/*
+ * bug-1886: re-applying a config updates the registered queue in place
+ * (the rejected duplicate used to be built, handed to the picker and
+ * leaked, while the registered queue kept its old settings), and the
+ * config it replaces is freed.
+ */
+static xce_scheduler_t g_reapply_sched;
+
+static void test_config_reapply_updates_queue_in_place_and_frees_old(void)
+{
+    xce_sched_plugin_t* fcfs = NULL;
+    xce_sched_queue_t*  q_first;
+    xce_sched_queue_t*  q_second;
+    xce_conf4sched_t*   first;
+    int                 rc;
+
+    memset(&g_reapply_sched, 0, sizeof(g_reapply_sched));
+    rc = xce_sched_core_init(&(g_reapply_sched.core));
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    fcfs = builtin_fcfs_new_plugin();
+    TEST_ASSERT_NOT_NULL(fcfs);
+    rc = plugin_register(&g_reapply_sched, fcfs);
+    TEST_ASSERT_TRUE(rc >= 0);
+    g_reapply_sched.core.job_pick_plugin = (char*)"builtin_fcfs";
+
+    first   = apply_queue_config(&g_reapply_sched, 1);
+    q_first = sched_get_queue(&(g_reapply_sched.core), QUEUE_NAME);
+    TEST_ASSERT_NOT_NULL(q_first);
+    TEST_ASSERT_EQUAL_INT(1, q_first->priority);
+
+    (void)apply_queue_config(&g_reapply_sched, 7);
+    q_second = sched_get_queue(&(g_reapply_sched.core), QUEUE_NAME);
+    TEST_ASSERT_EQUAL_PTR(q_first, q_second);
+    TEST_ASSERT_EQUAL_INT_MESSAGE(
+        7, q_second->priority,
+        "a re-applied config must update the registered queue in place");
+
+#if defined(__SANITIZE_ADDRESS__)
+    TEST_ASSERT_TRUE_MESSAGE(__asan_address_is_poisoned(first),
+                             "the replaced config must be freed");
+#else
+    (void)first;
+#endif
+}
+
 int main(void)
 {
     UNITY_BEGIN();
     RUN_TEST(test_config_push_mid_session_does_not_end_round);
     RUN_TEST(test_applied_config_push_is_recorded);
+    RUN_TEST(test_config_reapply_updates_queue_in_place_and_frees_old);
     return UNITY_END();
 }