From 154a169ff3138a2d8e5bbcfc96929a973b430742 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 27 Sep 2026 06:05:26 +0000 Subject: [PATCH] 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) --- src/controller/config_repl.c | 5 + src/controller/include/cluster.h | 6 + src/controller/meson.build | 1 + src/controller/persist.c | 8 ++ src/controller/queue.c | 53 ++++++++- src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_config_queue_reconcile.c | 106 ++++++++++++++++++ src/include/persist.h | 7 ++ 8 files changed, 183 insertions(+), 5 deletions(-) create mode 100644 src/controller/testsuite/test_config_queue_reconcile.c diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 1d85f5f1..7bab70c4 100644 --- a/src/controller/config_repl.c +++ b/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; } diff --git a/src/controller/include/cluster.h b/src/controller/include/cluster.h index 977a6d8c..ff3f7067 100644 --- a/src/controller/include/cluster.h +++ b/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*); diff --git a/src/controller/meson.build b/src/controller/meson.build index 445ca06d..0f7b7fe7 100644 --- a/src/controller/meson.build +++ b/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', diff --git a/src/controller/persist.c b/src/controller/persist.c index 8a9e49dc..3e596967 100644 --- a/src/controller/persist.c +++ b/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; diff --git a/src/controller/queue.c b/src/controller/queue.c index 038c8633..0f74f3cf 100644 --- a/src/controller/queue.c +++ b/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; diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index d8ca7dea..7ed4fe79 100644 --- a/src/controller/testsuite/controller_test.c +++ b/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); diff --git a/src/controller/testsuite/test_config_queue_reconcile.c b/src/controller/testsuite/test_config_queue_reconcile.c new file mode 100644 index 00000000..40abe061 --- /dev/null +++ b/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 + +#include +#include +#include + +#include + +#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); +} diff --git a/src/include/persist.h b/src/include/persist.h index 2fbc75c7..8f95d428 100644 --- a/src/include/persist.h +++ b/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);