fix(controller): a queue added by a live config upgrade exists without a restart (bug-1469)

build_queues() only ran at boot, so a queue a live config upgrade added
had no runtime object in c->qmgr (and the scheduler/apiserver never heard
of it) until the controller restarted. The checkpoint-ordering half of
bug-1469 was fixed in 959b0cda; this is the remaining half.

- reconcile_config_queues(c, notify): creates every queue in the active
  config that has no runtime object, leaves existing ones untouched,
  returns the number created.
- config_activate_committed() calls it (with notify) after every
  successful activation -- verified-marker and no-marker paths alike.
- persist_notify_queue_updated_no_wal(): fires HOOK_DETAIL_QUEUE_UPDATE
  under the t_apply_seam_notifying guard, so the scheduler/apiserver are
  told while no WAL_QUEUE_CTRL is appended -- config queues are
  config-owned, built on every member from the committed config.
- build_queues() now delegates to it without notify, so boot and
  activation are idempotent in either order.

New test_config_queue_reconcile.c (RED before: the added queue was never
created): creates the added queue, leaves the existing object untouched,
notifies once, and a second reconcile/build_queues adds nothing.
Controller suite passes.

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

src/controller/config_repl.c

@@ -326,6 +326,8 @@ int config_activate_committed(cluster_t *c)
         if (lrc != 0 && lrc != -ENOENT)
             WARN("config_activate_committed: no committed marker; "
                  "load_primary_config fallback rc=%d", lrc);
+        if (lrc == 0)
+            reconcile_config_queues(c, 1); /* bug-1469 */
         return 0;
     }
     if (rc != 0)
@@ -373,6 +375,9 @@ int config_activate_committed(cluster_t *c)
     atomic_store_explicit(&c->ha.config_serve_fence, 0, memory_order_release);
     INFO("config activated from committed marker tag=%016" PRIx64
          " (hash-verified out-of-band bytes)", marker_tag);
+    /* bug-1469: materialize queues the new config added (they used to need
+     * a restart -- build_queues() only ran at boot). */
+    reconcile_config_queues(c, 1);
     return 0;
 }
 

src/controller/include/cluster.h

@@ -78,6 +78,12 @@ extern int do_job_started(void*, uint64_t, int, int*, int, int*);
 
 extern queue_t* create_queue(cluster_t*, xce_queue_t*);
 extern int      build_queues(void*);
+/* bug-1469: create every queue in the active config that has no runtime
+ * object yet (existing ones are left untouched); with `notify`, tell the
+ * scheduler/apiserver about each new queue without appending a WAL fact
+ * (config queues are config-owned, built on every member). Returns the
+ * number created. */
+extern int reconcile_config_queues(cluster_t*, int notify);
 extern void     queue_fini(queue_t*);
 extern int      queue_ctrl(cluster_t*, xce_queuectrl_t*);
 

src/controller/meson.build

@@ -154,6 +154,7 @@ controller_unit_test_src = [
   'testsuite/test_wal_job_start_fact.c',
   'testsuite/test_wal_job_finish_fact.c',
   'testsuite/test_wal_job_fail_fact.c',
+  'testsuite/test_config_queue_reconcile.c',
   'testsuite/test_wal_job_rusage_fact.c',
   'testsuite/test_wal_job_ckpt_fact.c',
   'testsuite/test_wal_job_preason_fact.c',

src/controller/persist.c

@@ -8192,6 +8192,14 @@ void persist_notify_job_updated_already_persisted(cluster_t* c, job_t* j)
  * status changed WITHOUT persist_worker_hook_fun appending a WAL_WORKER_CTRL
  * fact for it, since resync.c's XCE_WORKER_RECONCILING coordination bit is
  * intentionally never durable. */
+void persist_notify_queue_updated_no_wal(cluster_t* c, queue_t* q)
+{
+    t_apply_seam_notifying = true;
+    xce_run_hooks(&c->hooksys, XCE_CONTROLLER_HOOK_QUEUE, HOOK_AFTER_EVENT,
+                  HOOK_DETAIL_QUEUE_UPDATE, c, q);
+    t_apply_seam_notifying = false;
+}
+
 void persist_notify_worker_status_updated(cluster_t* c, worker_t* w)
 {
     t_apply_seam_notifying = true;

src/controller/queue.c

@@ -38,15 +38,58 @@ int build_queues(void* c)
         return -EINVAL;
     }
 
-    for (i = 0; i < cls->myconf->cluster_config->nqueues; i++) {
-        xce_queue_t* q = cls->myconf->cluster_config->queues[i];
-
-        create_queue(cls, q);
-    }
+    /* bug-1469: idempotent -- skips queues that already exist, so boot and
+     * config activation (reconcile_config_queues with notify) can run in
+     * either order. */
+    (void)i;
+    reconcile_config_queues(cls, 0);
 
     return 0;
 }
 
+/* bug-1469: see cluster.h. */
+int reconcile_config_queues(cluster_t* c, int notify)
+{
+    int created = 0;
+    int i;
+
+    if ((c == NULL) || (c->qmgr == NULL) || (c->myconf == NULL)
+        || (c->myconf->cluster_config == NULL)) {
+        return 0;
+    }
+
+    for (i = 0; i < c->myconf->cluster_config->nqueues; i++) {
+        xce_queue_t* qdef = c->myconf->cluster_config->queues[i];
+        queue_t*     q    = NULL;
+
+        if ((qdef == NULL) || (qdef->queue == NULL)) {
+            continue;
+        }
+
+        lock(c->qmgr, RTTI_LOCK_READ);
+        key_find(c->qmgr, (void*)qdef->queue, (void**)&q);
+        unlock(c->qmgr);
+        if (q != NULL) {
+            continue; /* existing runtime queue: left untouched */
+        }
+
+        q = create_queue(c, qdef);
+        if (q == NULL) {
+            WARN("%s: failed to create queue %s from configuration", __func__,
+                 qdef->queue);
+            continue;
+        }
+        created++;
+
+        if (notify) {
+            INFO("queue %s added by configuration activation", qdef->queue);
+            persist_notify_queue_updated_no_wal(c, q);
+        }
+    }
+
+    return created;
+}
+
 int queue_ctrl(cluster_t* c, xce_queuectrl_t* ctrl)
 {
     queue_t* q  = NULL;

src/controller/testsuite/controller_test.c

@@ -85,6 +85,7 @@ extern void test_wal_job_dispatch_fact(void);
 extern void test_wal_job_start_fact(void);
 extern void test_wal_job_finish_fact(void);
 extern void test_wal_job_fail_fact(void);
+extern void test_config_queue_reconcile(void);
 extern void test_wal_job_rusage_fact(void);
 extern void test_wal_job_ckpt_fact(void);
 extern void test_wal_job_preason_fact(void);
@@ -271,6 +272,7 @@ int main(int argc, char* argv[])
     TEST(test_wal_job_start_fact);
     TEST(test_wal_job_finish_fact);
     TEST(test_wal_job_fail_fact);
+    TEST(test_config_queue_reconcile);
     TEST(test_wal_job_rusage_fact);
     TEST(test_wal_job_ckpt_fact);
     TEST(test_wal_job_preason_fact);

src/controller/testsuite/test_config_queue_reconcile.c

@@ -0,0 +1,106 @@
+/* test_config_queue_reconcile.c -- bug-1469: a queue added by a live config
+ * upgrade used to exist only after a restart, because build_queues() ran
+ * only at boot. reconcile_config_queues() creates the missing runtime queue
+ * objects when a config is activated, notifies the scheduler/apiserver
+ * (without a WAL fact -- config queues are config-owned), leaves existing
+ * queues untouched, and is idempotent (boot's build_queues() uses it too). */
+#include <rtti.h>
+
+#include <stdarg.h>
+#include <stdio.h>
+#include <string.h>
+
+#include <unity/unity.h>
+
+#include "clusterconfig.h"
+#include "cluster.h"
+#include "hook.h"
+#include "myconf.h"
+#include "qobj.h"
+#include "xce/cluster_def.h"
+#include "xstring.h"
+
+static int g_cqr_queue_updates;
+
+static int cqr_queue_spy(int type, int situation, int detail, void* data,
+                         va_list ap)
+{
+    (void)type; (void)data; (void)ap;
+    if (situation == HOOK_AFTER_EVENT && detail == HOOK_DETAIL_QUEUE_UPDATE)
+        g_cqr_queue_updates++;
+    return 0;
+}
+
+static queue_t* cqr_find(cluster_t* c, const char* name)
+{
+    queue_t* q = NULL;
+    lock(c->qmgr, RTTI_LOCK_READ);
+    key_find(c->qmgr, (void*)name, (void**)&q);
+    unlock(c->qmgr);
+    return q;
+}
+
+static void test_reconcile_creates_queue_added_by_config(void)
+{
+    kvmgr_params_t   str_params = { .key_type = KVMGR_KEY_STRING,
+                                    .nslots = 16, .delta = 8 };
+    cluster_t        c;
+    myconf_t         conf;
+    cluster_config_t cfg;
+    xce_queue_t      qa, qb;
+    xce_queue_t*     queues[2];
+    queue_t*         before;
+
+    if (makeString((char*)"cqr-probe") == NULL)
+        xstring_table_init(64);
+
+    memset(&c, 0, sizeof(c));
+    memset(&conf, 0, sizeof(conf));
+    memset(&cfg, 0, sizeof(cfg));
+    TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&c.hooksys, XCE_CONTROLLER_HOOK_MAX));
+    c.qmgr = makeInstance(&xkvmgr_type, &str_params);
+    TEST_ASSERT_NOT_NULL(c.qmgr);
+
+    memset(&qa, 0, sizeof(qa));
+    qa.queue = (char*)"cqr-existing";
+    memset(&qb, 0, sizeof(qb));
+    qb.queue = (char*)"cqr-added";
+    queues[0] = &qa;
+    queues[1] = &qb;
+    cfg.nqueues = 2;
+    cfg.queues  = queues;
+    conf.cluster_config = &cfg;
+    c.myconf = &conf;
+
+    /* The old config's queue already exists at runtime. */
+    TEST_ASSERT_NOT_NULL(create_queue(&c, &qa));
+    before = cqr_find(&c, "cqr-existing");
+    TEST_ASSERT_NOT_NULL(before);
+
+    g_cqr_queue_updates = 0;
+    TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&c.hooksys, XCE_CONTROLLER_HOOK_QUEUE,
+                                          cqr_queue_spy, NULL));
+
+    TEST_ASSERT_EQUAL_INT_MESSAGE(1, reconcile_config_queues(&c, 1),
+        "bug-1469: the queue added by the new config must be created");
+    TEST_ASSERT_NOT_NULL(cqr_find(&c, "cqr-added"));
+    TEST_ASSERT_EQUAL_PTR_MESSAGE(before, cqr_find(&c, "cqr-existing"),
+        "an existing queue object must be left untouched");
+    TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_cqr_queue_updates,
+        "the scheduler/apiserver must be told about the new queue");
+
+    /* Idempotent: a second activation (or boot's build_queues) adds nothing. */
+    TEST_ASSERT_EQUAL_INT(0, reconcile_config_queues(&c, 1));
+    TEST_ASSERT_EQUAL_INT(1, g_cqr_queue_updates);
+    TEST_ASSERT_EQUAL_INT(0, build_queues(&c));
+    TEST_ASSERT_EQUAL_PTR(before, cqr_find(&c, "cqr-existing"));
+
+    xce_hook_fini(&c.hooksys);
+    dereference(c.qmgr);
+}
+
+void test_config_queue_reconcile(void)
+{
+    UnitySetTestFile(__FILE__);
+    RUN_TEST(test_reconcile_creates_queue_added_by_config);
+}

src/include/persist.h

@@ -1578,6 +1578,13 @@ void persist_notify_job_updated_already_persisted(cluster_t* c, job_t* j);
  * them until some unrelated worker event happens to re-fire the hook. */
 void persist_notify_worker_status_updated(cluster_t* c, worker_t* w);
 
+/* bug-1469: fire HOOK_DETAIL_QUEUE_UPDATE AFTER_EVENT for `q` under the same
+ * t_apply_seam_notifying guard, so the scheduler/apiserver learn about a
+ * queue created by config activation while persist_queue_hook_fun appends
+ * no WAL_QUEUE_CTRL fact -- config queues are config-owned and built on
+ * every member from the committed config, never replayed from WAL. */
+void persist_notify_queue_updated_no_wal(cluster_t* c, queue_t* q);
+
 int persist_worker_hook_fun(int type, int situation, int detail,
                             void* cls, va_list ap);