lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

admin-plane breakpoint 1: config activation reaches live scheduler + queues hot-apply

Browse source 
Three fixes so a pushed config edit takes effect on the RUNNING cluster
instead of waiting for a scheduler rejoin (admin-plane-gaps 1.1):

1. queue_refresh_def (controller/queue.c): reconcile_config_queues no
   longer leaves existing runtime queues untouched -- it refreshes their
   CONFIG-owned fields in place (resreq/pre_exec/post_exec interned
   swaps; users/hosts/thresholds/windows via free_* + MOVE_PTR, same
   ownership as queue_init; scalars copied) and notifies the live
   scheduler/apiserver via the same NO-WAL hook a fresh queue gets.

2. config_refresh (controller/cluster.c): cluster_build_config is now
   idempotent -- boot creates the confvar entries, activation re-runs
   refresh their interned values in place (makeConfig refuses EEXIST,
   kvmgr has no del_key), so heartbeat_interval/
   inactive_heartbeat_count exports stop being frozen at boot values.

3. activation re-push (controller/config_repl.c): config_activate_
   committed now calls cluster_build_config then xcesched_push_configs
   (un-static'd) to the connected scheduler, on both the hash-verified
   and -ENOENT fallback activation paths. Lock order follows the
   established apply_pump_lock -> sched.lock seam; push failure only
   WARNs (scheduler keeps its previous config until next join/push).

E2E-found display gap: apiserver's Go handleNotifyData had no
NotifyData_data_schedUpdateQueue case, so the controller's per-queue
notify was silently dropped and `xce queues list` kept the join-time
snapshot. Added the case (OnQueueUpdated) so hot-applied priority and
queue state changes are visible immediately.

Unit tests: reconcile refresh-in-place + notify counts (test_config_
queue_reconcile.c); activation repushes once per activation, not while
deferred (test_config_repl.c capturing fake); OnQueueUpdated in-place
refresh (handler_test.go). Full controller suite green.

E2E on t1/t2/t3: two config pushes (gen 8->9->10) with no restarts --
controller activation + per-queue refresh + scheduler re-push logged,
scheduler applied at the round boundary, `xce queues list` showed the
new priority immediately, and a submit->RUN->DONE job round-tripped on
the hot-edited queue with all four daemons healthy.

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
lei.chen 2026-10-23 09:24:05 +00:00
parent d16954ad10
commit 563f8e6890
10 changed files with 743 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

55
src/controller/cluster.c
View file

@ -1463,6 +1463,59 @@ static void endpoint_killer(void* p)
return;
}
/* config_refresh: update an EXISTING configmgr entry's value in place.
* makeConfig refuses EEXIST (configobj.c), so re-running cluster_build_config
* after boot could never refresh the confvars the scheduler/worker config
* pushes export -- they would stay frozen at their boot-time values even
* after a config activation changed them (admin-plane-gaps 1.1: config push
* never carrying the activated values). key_find the entry under the
* configmgr WRITE lock and swap the interned value string (dereference old,
* makeString new -- xce_config_t.value is an interned xstring, same
* ownership convention qobj.c/jobj.c use for resreq).
*
* Caller must NOT hold c->configmgr. Returns 0 on refresh (or "entry absent
* -- nothing to refresh"), negative on a real error. */
static int config_refresh(cluster_t* c, const char* name, const char* value)
{
xce_config_t* cv = NULL;
int rc;
if (c->configmgr == NULL) {
/* No config manager (bare/test fixture) -- nothing to refresh;
* add_config's caller owns creation on that path too. */
return 0;
}
lock(c->configmgr, RTTI_LOCK_WRITE);
rc = key_find(c->configmgr, (void*)name, (void**)&cv);
if (rc > 0 && cv != NULL) {
/* bug-447 precedent: binary key_find returns 1 on found. */
char* nv = makeString((char*)value);
if (nv == NULL) {
unlock(c->configmgr);
return -ENOMEM;
}
if (cv->value != NULL) {
dereference(cv->value);
}
cv->value = nv;
rc = 0;
} else if (rc == 0 || rc == -ENOENT) {
rc = 0; /* not built yet -- add_config's caller owns creation */
}
/* else: key_find already returns a negative errno -- keep it */
unlock(c->configmgr);
return rc;
}
/* cluster_build_config is idempotent: boot (xce_config_init's
* HOOK_DETAIL_CLUSTER_INIT path) creates the confvar entries via add_config;
* a re-run after a config activation (config_repl.c's activated-seam call)
* REFRESHES them instead of failing on EEXIST. The confvars are what
* collect_confvars() exports to the scheduler/worker on every config push,
* so without this refresh a live config change to heartbeat_interval/
* inactive_heartbeat_count would never reach a running scheduler. */
int cluster_build_config(cluster_t* c)
{
char valbuf[128];
@ -1470,9 +1523,11 @@ int cluster_build_config(cluster_t* c)
snprintf(valbuf, sizeof(valbuf) - 1, "%d",
c->myconf->cluster_config->heartbeat_interval);
add_config(c, "heartbeat_interval", valbuf);
(void)config_refresh(c, "heartbeat_interval", valbuf);
snprintf(valbuf, sizeof(valbuf) - 1, "%d",
c->myconf->cluster_config->inactive_heartbeat_count);
add_config(c, "inactive_heartbeat_count", valbuf);
(void)config_refresh(c, "inactive_heartbeat_count", valbuf);
return 0;
}

63
src/controller/config_repl.c
View file

@ -335,8 +335,19 @@ 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)
if (lrc == 0) {
reconcile_config_queues(c, 1); /* bug-1469 */
/* admin-plane-gaps 1.1: same confvar refresh + live-scheduler
* re-push as the hash-verified activation path below (this
* fallback genuinely activates locally-held bytes on recovery,
* so the same materialization applies). */
(void)cluster_build_config(c);
mylock_lock(&(c->sched.lock));
if (c->sched.ep != NULL) {
(void)xcesched_push_configs(c);
}
mylock_unlock(&(c->sched.lock));
}
return 0;
}
if (rc != 0)
@ -387,6 +398,56 @@ int config_activate_committed(cluster_t *c)
/* 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);
/* admin-plane-gaps 1.1 (config reachability): materialize the rest of
* the activated configuration into live runtime state and re-push it to
* the connected scheduler, so a running scheduler picks the change up
* WITHOUT re-joining:
* 1. cluster_build_config re-runs its boot-time confvar seeding, now
* refresh-in-place (config_refresh, cluster.c) -- the confvars the
* scheduler/worker config pushes export (heartbeat_interval/
* inactive_heartbeat_count) were previously frozen at their
* boot-time values.
* 2. reconcile_config_queues above already refreshes EXISTING queues
* in place (queue.c's queue_refresh_def) and notifies the
* scheduler/apiserver per updated/created queue.
* 3. xcesched_push_configs sends the full conf4sched (confvars,
* queues, resources, shared resvals, ugroups/hgroups/limits/
* hproviders) to the live scheduler. The receive side is fully
* hot-apply capable (handler.c -> sched_stash_config -> main.c's
* round-boundary sched_apply_pending_config -> sched_init_config's
* in-place queue update, bug-1886); only this controller-side
* trigger was missing -- the push used to happen on scheduler join
* ONLY (sched.c on_scheduler_join), so a running scheduler kept
* its boot-time config until it re-joined.
*
* Lock ordering: this call runs while persist_apply_pump_run holds
* apply_pump_lock (the established WAL_CONFIG_MARKER apply seam, see
* the task-5 comment above). Taking c->sched.lock here follows the
* SAME apply_pump_lock -> sched.lock order this seam already
* exercises (the WAL_QUEUE_CTRL case's hook fire reaches sched_queue_
* update, which takes c->sched.lock under the same held apply_pump_
* lock), so no new lock-order edge and no reverse edge exists
* (nothing holding c->sched.lock ever acquires apply_pump_lock --
* xce_pushconfig_req only enqueues on the endpoint outq). The push
* itself is a non-blocking enqueue, the same I/O class this seam
* already performs. */
(void)cluster_build_config(c);
mylock_lock(&(c->sched.lock));
if (c->sched.ep != NULL) {
int prc = xcesched_push_configs(c);
if (prc < 0) {
WARN("config activation tag=%016" PRIx64 ": scheduler config "
"push failed rc=%d -- scheduler keeps its previous config "
"until its next join or config push", marker_tag, prc);
} else {
INFO("config activation tag=%016" PRIx64 ": re-pushed configs "
"to the connected scheduler", marker_tag);
}
}
mylock_unlock(&(c->sched.lock));
return 0;
}

24
src/controller/include/cluster.h
View file

@ -33,6 +33,11 @@ extern int cluster_start(cluster_t*);
extern int cluster_run(cluster_t*);
extern void cluster_fini(cluster_t*);
extern int register_types(cluster_t*);
/* admin-plane-gaps 1.1: create (boot) or refresh (config activation) the
* confvar entries the scheduler/worker config pushes export. Idempotent --
* safe to call from both the boot hook and config_activate_committed's
* activated seam. */
extern int cluster_build_config(cluster_t*);
extern int worker_ctrl(cluster_t*, xce_hostctrl_t*);
extern xce_resource_t* get_resource(cluster_t*, char*);
extern int add_to_xlist(xlist_t*, void*);
@ -101,17 +106,28 @@ 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. */
* object yet; with `notify`, tell the scheduler/apiserver about each
* created-or-refreshed queue without appending a WAL fact (config queues
* are config-owned, built on every member). admin-plane-gaps 1.1: an
* EXISTING runtime queue is no longer left untouched -- its config-owned
* fields are refreshed in place (queue_refresh_def) so a config activation
* that edits a queue hot-applies on the controller too. Returns the
* number created (refreshed queues are reported via TRACE, not the return). */
extern int reconcile_config_queues(cluster_t*, int notify);
/* admin-plane-gaps 1.1: (re)build/refresh the confvar entries the
* scheduler/worker config pushes export. Idempotent -- creates on first
* run (boot), refreshes in place on later runs (config activation). */
extern int cluster_build_config(cluster_t*);
extern void queue_fini(queue_t*);
extern int queue_ctrl(cluster_t*, xce_queuectrl_t*);
extern int xce_update_data(endpoint_t*, cluster_t*, int, int, int*, int,
xce_kv_t*);
extern int on_scheduler_join(cluster_t*);
/* admin-plane-gaps 1.1: (re)send the full conf4sched config push to the
* connected scheduler. Caller must hold c->sched.lock. No-op (0) when no
* cluster config is loaded yet. */
extern int xcesched_push_configs(cluster_t*);
extern int on_apiserver_join(cluster_t*, endpoint_t*);
extern int xce_push_reslist(endpoint_t*, cluster_t*, int, xce_resource_t**);

135
src/controller/queue.c
View file

@ -47,10 +47,117 @@ int build_queues(void* c)
return 0;
}
/* bug-1469: see cluster.h. */
/* queue_refresh_def: apply a queue-definition's CONFIG-owned fields to an
* existing runtime queue, in place (the controller-side twin of the
* scheduler's bug-1886 in-place re-apply in sched_init_queues()/
* core_upsert_queue()). Called under `q` RTTI_LOCK_WRITE.
*
* Ownership notes (mirrors qobj.c's queue_init, which built these fields
* originally):
* - resreq/pre_exec/post_exec are interned xstrings -- dereference the
* old, makeString the new (never free()).
* - users/hosts/thresholds/windows are decoded capnp payloads owned by
* the queue -- free with the generated free_* helpers, replace with
* the NEW definition's payloads via MOVE_PTR (the config's
* cluster_config owns the old copies and is retired by
* load_primary_config after activation; the runtime queue must not
* alias into it).
* - status/priority/accept_interval/sched_delay/attrs are scalars --
* plain copies (config is the source of truth; a re-activation that
* changed an admin state bit in the config applies it, and later WAL
* re-applies of queue_ctrl facts still win on replay order).
*
* Returns 0, or -ENOMEM when an interned-string allocation failed (the
* queue keeps its previous definition's strings in that case -- scalars
* may already have been copied, which is benign: the next activation
* re-applies). */
static int queue_refresh_def(queue_t* q, xce_queue_t* qdef)
{
char* ns;
ns = makeString(qdef->resreq);
if (ns == NULL && qdef->resreq != NULL) {
return -ENOMEM;
}
if (q->queue.resreq != NULL) {
dereference(q->queue.resreq);
}
q->queue.resreq = ns;
q->queue.attrs = qdef->attrs;
q->queue.status = qdef->status;
q->queue.priority = qdef->priority;
q->queue.accept_interval = qdef->accept_interval;
q->queue.sched_delay = qdef->sched_delay;
if (q->queue.users != NULL) {
free_GroupConfig_ptr(&(q->queue.users));
}
MOVE_PTR(q->queue.users, qdef->users);
if (q->queue.hosts != NULL) {
free_GroupConfig_ptr(&(q->queue.hosts));
}
MOVE_PTR(q->queue.hosts, qdef->hosts);
if (q->queue.nthresholds > 0 && q->queue.thresholds != NULL) {
free_Threshold_list(q->queue.nthresholds, q->queue.thresholds);
q->queue.thresholds = NULL;
}
q->queue.nthresholds = qdef->nthresholds;
MOVE_PTR(q->queue.thresholds, qdef->thresholds);
if (q->queue.runwindow_config != NULL) {
free_TimeWindow_ptr(&(q->queue.runwindow_config));
}
MOVE_PTR(q->queue.runwindow_config, qdef->runwindow_config);
if (q->queue.dispatchwindow_config != NULL) {
free_TimeWindow_ptr(&(q->queue.dispatchwindow_config));
}
MOVE_PTR(q->queue.dispatchwindow_config, qdef->dispatchwindow_config);
ns = makeString(qdef->pre_exec);
if (ns == NULL && qdef->pre_exec != NULL) {
return -ENOMEM;
}
if (q->queue.pre_exec != NULL) {
dereference(q->queue.pre_exec);
}
q->queue.pre_exec = ns;
ns = makeString(qdef->post_exec);
if (ns == NULL && qdef->post_exec != NULL) {
return -ENOMEM;
}
if (q->queue.post_exec != NULL) {
dereference(q->queue.post_exec);
}
q->queue.post_exec = ns;
ns = makeString(qdef->autopscaler);
if (ns == NULL && qdef->autopscaler != NULL) {
return -ENOMEM;
}
if (q->queue.autopscaler != NULL) {
dereference(q->queue.autopscaler);
}
q->queue.autopscaler = ns;
return 0;
}
/* bug-1469: see cluster.h. Admin-plane-gaps 1.1 follow-up: an existing
* runtime queue is no longer "left untouched" on a config activation --
* its CONFIG-owned fields are refreshed in place (queue_refresh_def) and,
* with `notify`, the live scheduler/apiserver hear about the change via
* the same NO-WAL queue-update hook a fresh queue gets (config queues are
* config-owned, so no WAL_QUEUE_CTRL fact is appended for the refresh).
* Creation semantics are unchanged. */
int reconcile_config_queues(cluster_t* c, int notify)
{
int created = 0;
int updated = 0;
int i;
if ((c == NULL) || (c->qmgr == NULL) || (c->myconf == NULL)
@ -70,7 +177,26 @@ int reconcile_config_queues(cluster_t* c, int notify)
key_find(c->qmgr, (void*)qdef->queue, (void**)&q);
unlock(c->qmgr);
if (q != NULL) {
continue; /* existing runtime queue: left untouched */
int rc;
/* existing runtime queue: refresh its config-owned fields */
lock(q, RTTI_LOCK_WRITE);
rc = queue_refresh_def(q, qdef);
unlock(q);
if (rc < 0) {
WARN("%s: failed to refresh queue %s from configuration "
"(rc=%d) -- keeping previous definition",
__func__, qdef->queue, rc);
continue;
}
updated++;
if (notify) {
INFO("queue %s updated by configuration activation",
qdef->queue);
persist_notify_queue_updated_no_wal(c, q);
}
continue;
}
q = create_queue(c, qdef);
@ -87,6 +213,11 @@ int reconcile_config_queues(cluster_t* c, int notify)
}
}
if (updated > 0) {
TRACE("%s: %d queue(s) refreshed from configuration", __func__,
updated);
}
return created;
}

12
src/controller/sched.c
View file

@ -136,7 +136,17 @@ static int xcesched_push_workerlist(cluster_t* c, xlist_t* l)
return rc;
}
static int xcesched_push_configs(cluster_t* c)
/* xcesched_push_configs: build+send the scheduler's conf4sched config push
* (confvars/queues/resources/shared resvals from the runtime managers +
* ugroups/hgroups/limits/hproviders from the active cluster config).
* admin-plane-gaps 1.1: no longer static -- config_repl.c's activation seam
* re-pushes configs to a CONNECTED scheduler after a config activation
* (the receive side, sched_stash_config -> sched_apply_pending_config ->
* sched_init_config, is fully hot-apply capable; only the controller-side
* trigger was missing, so a running scheduler kept its boot-time config
* until it re-joined). Caller must hold c->sched.lock, exactly like
* on_scheduler_join's push branch. */
int xcesched_push_configs(cluster_t* c)
{
xce_configdata_t configdata;
int rc = 0;

96
src/controller/testsuite/test_config_queue_reconcile.c
View file

@ -2,8 +2,11 @@
* 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). */
* (without a WAL fact -- config queues are config-owned). admin-plane-gaps
* 1.1: an EXISTING queue is no longer left untouched -- its config-owned
* fields are refreshed in place (queue_refresh_def) and it gets the same
* no-WAL notify a created queue does; and reconcile stays idempotent (boot's
* build_queues() uses it too). */
#include <rtti.h>
#include <stdarg.h>
@ -81,17 +84,21 @@ static void test_reconcile_creates_queue_added_by_config(void)
TEST_ASSERT_EQUAL_INT(0, xce_add_hook(&c.hooksys, XCE_CONTROLLER_HOOK_QUEUE,
cqr_queue_spy, NULL));
/* 1 created + 1 refreshed, each with its own notify fire. */
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");
"an existing queue must be refreshed in place, not replaced");
TEST_ASSERT_EQUAL_INT_MESSAGE(2, g_cqr_queue_updates,
"the scheduler/apiserver must hear about the created AND the "
"refreshed queue (admin-plane-gaps 1.1)");
/* Idempotent: a second activation (or boot's build_queues) adds nothing. */
/* Idempotent: a second activation (or boot's build_queues) adds nothing;
* the refresh re-applies the same definition to BOTH queues (2 more
* notifies), so still no new object. */
TEST_ASSERT_EQUAL_INT(0, reconcile_config_queues(&c, 1));
TEST_ASSERT_EQUAL_INT(1, g_cqr_queue_updates);
TEST_ASSERT_EQUAL_INT(4, g_cqr_queue_updates);
TEST_ASSERT_EQUAL_INT(0, build_queues(&c));
TEST_ASSERT_EQUAL_PTR(before, cqr_find(&c, "cqr-existing"));
@ -99,8 +106,83 @@ static void test_reconcile_creates_queue_added_by_config(void)
dereference(c.qmgr);
}
/* admin-plane-gaps 1.1: editing an EXISTING queue in the activated config
* must hot-apply onto the live runtime queue in place: scalar fields
* (status/priority/...) copied, the interned resreq string swapped, and the
* same queue_t object kept (never destroyed + recreated -- dispatchers hold
* that pointer). */
static void test_reconcile_refreshes_existing_queue_in_place(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;
xce_queue_t* queues[1];
queue_t* q;
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-edit";
qa.status = 1;
qa.priority = 10;
qa.resreq = (char*)"oldres=1";
qa.pre_exec = (char*)"old_pre.sh";
queues[0] = &qa;
cfg.nqueues = 1;
cfg.queues = queues;
conf.cluster_config = &cfg;
c.myconf = &conf;
TEST_ASSERT_NOT_NULL(create_queue(&c, &qa));
q = cqr_find(&c, "cqr-edit");
TEST_ASSERT_NOT_NULL(q);
TEST_ASSERT_EQUAL_INT(10, q->queue.priority);
TEST_ASSERT_EQUAL_STRING("oldres=1", q->queue.resreq);
/* The admin edits the queue in the new config version. The config object
* itself is the "activated" one here, so edit it in place (a real
* activation swaps in a fresh cluster_config that reconcile then reads;
* queue_refresh_def reads the definition but only takes ownership of the
* capnp payloads, which this scalar/string-focused test leaves NULL). */
qa.status = 0;
qa.priority = 42;
qa.resreq = (char*)"newres=2";
qa.pre_exec = (char*)"new_pre.sh";
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(0, reconcile_config_queues(&c, 1),
"an edited queue must refresh, not create (return counts creations)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_cqr_queue_updates,
"the refreshed queue must be announced to the scheduler/apiserver");
/* Same object, new values -- the in-place hot-apply contract. */
TEST_ASSERT_EQUAL_PTR(q, cqr_find(&c, "cqr-edit"));
TEST_ASSERT_EQUAL_INT(0, q->queue.status);
TEST_ASSERT_EQUAL_INT(42, q->queue.priority);
TEST_ASSERT_EQUAL_STRING("newres=2", q->queue.resreq);
TEST_ASSERT_EQUAL_STRING("new_pre.sh", q->queue.pre_exec);
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);
RUN_TEST(test_reconcile_refreshes_existing_queue_in_place);
}

137
src/controller/testsuite/test_config_repl.c
View file

@ -1096,6 +1096,10 @@ static void t43_commit_marker(cluster_t *c, uint64_t tag, const uint8_t sha[32])
t42_wait_applied(&c->persist, mlsn, 2000);
}
/* Capturing fake for xcesched_push_configs (defined at file bottom with the
* other cluster.c-link fakes; sched.c is deliberately not linked here). */
extern int g_crp_push_configs_called;
/* (4.3-a) MATCH activates: a committed marker whose locally-held bytes' sha256
* MATCHES the marker activates the configuration and leaves the member UN-fenced
* (R7.8). */
@ -1139,6 +1143,122 @@ static void test_config_activate_match_activates_and_unfences(void)
printf("test_config_activate_match_activates_and_unfences: PASS\n");
}
/* (4.3-a2) admin-plane-gaps 1.1: a hash-verified activation re-pushes the
* full conf4sched to a CONNECTED scheduler (xcesched_push_configs), so a
* running scheduler picks the new config up WITHOUT re-joining; and the
* push must NOT fire when activation is deferred (-EAGAIN, bytes
* mismatch/absent) -- the fence's own retry path is what drives activation
* later. Observed via the capturing xcesched_push_configs fake (above).
* c->sched.ep is wired to a zeroed fd=-1 stack endpoint (same fixture shape
* test_job_array_sched_push.c proves safe: packets queue on ep->outq
* without touching a socket; the fake never even looks at it).
*
* Note the apply pump ALREADY calls config_activate_committed during
* t43_commit_marker (apply_committed_entry's WAL_CONFIG_MARKER case), while
* sched.ep is still NULL -- so the marker-commit-driven fire does not count;
* only this test's own explicit, endpoint-wired call must. */
static void test_config_activation_repushes_configs_to_connected_scheduler(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_crp11_push_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
cluster_t *c = t42_make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c);
t42_make_leader(c, /*term*/3, /*n_peers*/1);
TEST_ASSERT_EQUAL_INT(0,
commit_engine_init(&c->commit, &c->persist, 1, 8));
repl_set_commit_engine(&c->persist.repl, &c->commit);
myconf_t *mc = t43_attach_myconf(c, tmp);
size_t blen = 0;
uint8_t *bytes = t43_make_config_blob(&blen);
uint8_t sha[32];
SHA256(bytes, blen, sha);
uint64_t tag;
{ uint64_t t; memcpy(&t, sha, 8); tag = be64toh(t); }
TEST_ASSERT_EQUAL_INT(0, conf_save_config(mc, bytes, blen, sha));
free(bytes);
/* The marker commit itself drives activation via the apply pump, while
* sched.ep is still NULL -- no push fires (and no crash from the NULL). */
g_crp_push_configs_called = 0;
t43_commit_marker(c, tag, sha);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_crp_push_configs_called,
"no scheduler push while the scheduler is not connected");
/* Wire a (zeroed, fd=-1) connected scheduler endpoint and activate. */
endpoint_t ep;
memset(&ep, 0, sizeof(ep));
ep.fd = -1;
c->sched.ep = &ep;
TEST_ASSERT_EQUAL_INT(0, config_activate_committed(c));
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_crp_push_configs_called,
"admin-plane-gaps 1.1: a hash-verified activation must re-push "
"configs to the connected scheduler");
TEST_ASSERT_EQUAL_INT(0, config_serve_fenced(c));
TEST_ASSERT_NOT_NULL(mc->cluster_config);
TEST_ASSERT_EQUAL_HEX64(tag, mc->config_version_tag);
/* A second activation (same committed marker, idempotent re-run) pushes
* again -- the push is unconditional on success, matching the
* on_scheduler_join push branch it mirrors. */
TEST_ASSERT_EQUAL_INT(0, config_activate_committed(c));
TEST_ASSERT_EQUAL_INT(2, g_crp_push_configs_called);
c->sched.ep = NULL;
t43_free_myconf(c);
commit_engine_fini(&c->commit);
t42_teardown(c);
printf("test_config_activation_repushes_configs_to_connected_scheduler: PASS\n");
}
/* (4.3-a3) admin-plane-gaps 1.1: the -EAGAIN defer path (committed marker's
* bytes absent/mismatched locally) must NOT push anything -- the member is
* fenced and waiting for bytes, so there is nothing activated to push. */
static void test_config_activation_deferred_does_not_push_configs(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_crp11_defer_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
cluster_t *c = t42_make_cluster(tmp);
TEST_ASSERT_NOT_NULL(c);
t42_make_leader(c, /*term*/4, /*n_peers*/1);
TEST_ASSERT_EQUAL_INT(0,
commit_engine_init(&c->commit, &c->persist, 1, 8));
repl_set_commit_engine(&c->persist.repl, &c->commit);
myconf_t *mc = t43_attach_myconf(c, tmp);
/* Commit a marker whose bytes were NEVER saved locally (the out-of-band
* delivery has not arrived) -- the classic fence-and-retry state. */
uint8_t shaB[32];
uint64_t tagB = t42_make_ref(shaB, 0x77);
g_crp_push_configs_called = 0;
t43_commit_marker(c, tagB, shaB);
endpoint_t ep;
memset(&ep, 0, sizeof(ep));
ep.fd = -1;
c->sched.ep = &ep;
TEST_ASSERT_EQUAL_INT_MESSAGE(-EAGAIN, config_activate_committed(c),
"absent bytes must defer activation (-EAGAIN), not push");
TEST_ASSERT_EQUAL_INT(1, config_serve_fenced(c));
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_crp_push_configs_called,
"a deferred (mismatched/absent-bytes) activation must NOT push "
"configs to the scheduler -- nothing was activated");
c->sched.ep = NULL;
t43_free_myconf(c);
commit_engine_fini(&c->commit);
t42_teardown(c);
printf("test_config_activation_deferred_does_not_push_configs: PASS\n");
}
/* (4.3-b) MISMATCH fences + retries, NEVER activates: an injected hash mismatch
* (locally-held bytes whose sha256 != the committed marker) keeps the member
* FENCED (non-serving), ALIVE, and retrying, and does NOT activate the mismatched
@ -3629,6 +3749,19 @@ int do_runtasks(cluster_t *c, endpoint_t *ep, xce_request_t *req)
int handle_apiserver_leave(cluster_t *c, endpoint_t *ep)
{ (void)c; (void)ep; return 0; }
/* admin-plane-gaps 1.1: config_repl.c's activation seam now calls
* xcesched_push_configs() (sched.c) -- which this binary deliberately does
* NOT link (same real-vs-fake collision the on_scheduler_join fake above
* documents). This capturing fake stands in for it: the new activation-push
* test below asserts on the call count, and every OTHER test in this file
* that drives config_activate_committed relies on the fake being a no-op
* (all t42/t43 fixtures leave c->sched.ep NULL, so the production guard
* skips the call entirely -- the counter exists purely for the one test
* that wires an endpoint in). */
int g_crp_push_configs_called;
int xcesched_push_configs(cluster_t *c)
{ (void)c; g_crp_push_configs_called++; return 0; }
static io_packet_t *cpr22_build_config_push_reply_pkt(int accepted,
uint64_t version_tag,
uint64_t term)
@ -7457,6 +7590,10 @@ void test_config_repl(void)
RUN_TEST(test_committed_marker_survives_checkpoint_trim);
RUN_TEST(test_config_activate_match_activates_and_unfences);
/* admin-plane-gaps 1.1: activation re-pushes configs to a connected
* scheduler; the -EAGAIN defer path pushes nothing. */
RUN_TEST(test_config_activation_repushes_configs_to_connected_scheduler);
RUN_TEST(test_config_activation_deferred_does_not_push_configs);
RUN_TEST(test_config_activate_mismatch_fences_and_does_not_activate);
RUN_TEST(test_config_activate_obtained_bytes_verified_before_activate);
RUN_TEST(test_config_activate_recovery_uses_latest_committed_marker);

14
src/frontend/apiserver/backend/clib.go
View file

@ -567,6 +567,20 @@ func handleNotifyData(nd *C.xce_notify_data_t) C.int {
}
gHandler.OnJobPidInfo(uint64(pi.jobid), cPidInfoToRaw(pi))
case C.NotifyData_data_schedUpdateQueue:
// Queue open/close/priority changes reach this API server as a
// per-queue notify (queue_ctrl's hook fire, and configuration
// activation's reconcile_config_queues refresh). Without this case
// the push was silently dropped here and store.Queues kept the
// join-time snapshot (dispatchConfigPush at OnBackendConnect) until
// the next full config push -- so `xce queues list` never reflected
// a hot-applied config edit or a queue state change.
qp := (*C.xce_queue_t)(C.xce_notify_data_ptr(nd))
if qp == nil {
return 0
}
gHandler.OnQueueUpdated(cQueueToModel(qp))
case C.NotifyData_data_schedUpdateResvals:
rvl := (*C.xce_resval_list_t)(C.xce_notify_data_ptr(nd))
if rvl == nil {

21
src/frontend/apiserver/backend/handler_test.go
View file

@ -448,6 +448,27 @@ func TestOnClusterUpdated_PartialMerge(t *testing.T) {
}
}
func TestOnQueueUpdated_InPlaceRefresh(t *testing.T) {
resetStores()
h := DefaultHandler{}
// Join-time config push seeded the queue table.
h.OnQueueUpdated(&models.QueueInfo{QueueName: "short", Priority: 45})
// A later per-queue notify (queue open/close or a hot-applied config
// activation) refreshes the SAME key in place — `xce queues list` must
// reflect the new priority without an apiserver restart.
h.OnQueueUpdated(&models.QueueInfo{QueueName: "short", Priority: 46})
got, ok := store.Queues.Get("short")
if !ok {
t.Fatal("queue short should exist in store after OnQueueUpdated")
}
if got.Priority != 46 {
t.Errorf("Priority: got %d, want 46 (notify must refresh in place)", got.Priority)
}
}
func TestOnHostUpdated_PartialMerge(t *testing.T) {
resetStores()
h := DefaultHandler{}