fix(scheduler): re-pushed resource definitions are not errors (bug-1854)

Every scheduler (re)join re-pushes the whole config, and both apply paths
-- sched_init_resources() (scheduler.c) and core_apply_config_resources()
(core_events.c) -- re-added each resource with makeResource(), which
refuses an existing name: one ERROR per resource per rejoin (33 in
data/xce_scheduler.log), and a genuinely changed definition was dropped
just as silently.

New libxce resourceApplyDef(): under one write lock, register a new
resource (ADDED), or compare an already-registered one on its
scheduling-relevant fields -- valtype, ordtype, flags, interval -- and
return SAME or CHANGED. Both call sites log SAME at debug, CHANGED as a
WARN that the definition is not applied to the running scheduler
(updating a registered resource in place would disturb its index and
accounting), and only real failures as ERROR.

Test test_resource_reapply_same_is_noop_changed_is_reported (EEXIST on
an identical re-push before). Scheduler suite 87/87, lib 1/1. Re-verify
of data/scheduler-replay-20260925-133708.replay: 0 add-resource ERRORs
(33 before), 0 changed-definition WARNs, 0 mismatches, same 60 DISPATCH /
908 PEND decisions.

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

src/include/resobj.h

@@ -8,4 +8,14 @@
 
 extern xce_resource_t* makeResource(var, xce_resource_t*);
 
+/* bug-1854: register a pushed resource definition, tolerating re-pushes.
+ * Returns RES_APPLY_ADDED (newly registered), RES_APPLY_SAME (already
+ * registered with the same scheduling-relevant definition: valtype,
+ * ordtype, flags, interval), RES_APPLY_CHANGED (registered with a different
+ * definition -- NOT applied; the registered one is kept), or -errno. */
+#define RES_APPLY_ADDED   1
+#define RES_APPLY_SAME    0
+#define RES_APPLY_CHANGED 2
+extern int resourceApplyDef(var, xce_resource_t*);
+
 #endif

src/lib/resobj.c

@@ -149,3 +149,40 @@ xce_resource_t* makeResource(var resmgr, xce_resource_t* r)
 
     return r1;
 }
+
+/* bug-1854: register a pushed resource definition, tolerating re-pushes.
+ * Every scheduler (re)join re-pushes the whole config; makeResource() refuses
+ * an existing name, so callers logged one ERROR per resource per rejoin and
+ * a genuinely changed definition was dropped just as silently. Only the
+ * scheduling-relevant fields are compared (idx/last_report are runtime
+ * state, desc is cosmetic); a changed definition is reported but NOT
+ * applied -- updating a registered resource in place would disturb its
+ * index and accounting. Lookup and add happen under one write lock. */
+int resourceApplyDef(var resmgr, xce_resource_t* r)
+{
+    xce_resource_t* cur = NULL;
+    int             rc;
+
+    if ((resmgr == NULL) || (r == NULL) || (r->name == NULL)) {
+        return -EINVAL;
+    }
+
+    lock(resmgr, RTTI_LOCK_WRITE);
+    rc = key_find(resmgr, (void*)r->name, (void**)&cur);
+    if ((rc > 0) && (cur != NULL)) {
+        int same = (cur->valtype == r->valtype) && (cur->ordtype == r->ordtype)
+                && (cur->flags == r->flags) && (cur->interval == r->interval);
+
+        unlock(resmgr);
+        return same ? RES_APPLY_SAME : RES_APPLY_CHANGED;
+    }
+    if ((rc < 0) && (rc != -ENOENT)) {
+        unlock(resmgr);
+        return rc;
+    }
+
+    cur = makeResource_(resmgr, r);
+    unlock(resmgr);
+
+    return (cur != NULL) ? RES_APPLY_ADDED : -errno;
+}

src/scheduler/src/core_events.c

@@ -417,16 +417,26 @@ static void core_apply_config_resources(xce_sched_core_t* core,
 
     for (i = 0; i < config->n_resources; i++) {
         xce_resource_t* r  = config->resources[i];
-        xce_resource_t* r1;
+        int             rc;
 
         if (r == NULL) {
             continue;
         }
 
-        r1 = makeResource(core->resmgr, r);
+        /* bug-1854: a re-push of an already-registered resource is the
-        if (r1 == NULL) {
+         * normal (re)join path, not an error */
-            ERROR("%s add resource %s failed", __func__,
+        rc = resourceApplyDef(core->resmgr, r);
+        if (rc == RES_APPLY_CHANGED) {
+            WARN("%s resource %s definition changed in the pushed config -- "
+                 "not applied to the running scheduler (restart to pick it "
+                 "up)",
+                 __func__, r->name ? r->name : "(null)");
+        } else if (rc == RES_APPLY_SAME) {
+            DBG("%s resource %s already registered, unchanged", __func__,
                   r->name ? r->name : "(null)");
+        } else if (rc < 0) {
+            ERROR("%s add resource %s failed (rc=%d)", __func__,
+                  r->name ? r->name : "(null)", rc);
         }
     }
 }

src/scheduler/src/scheduler.c

@@ -883,17 +883,26 @@ static int sched_init_resources(xce_scheduler_t* sched)
 
     for (i = 0; i < config->n_resources; i++) {
         xce_resource_t* r  = config->resources[i];
-        xce_resource_t* r1 = NULL;
+        int             rc;
 
         if (r == NULL) {
             continue;
         }
 
-        r1 = makeResource(sched->core.resmgr, r);
+        /* bug-1854: a re-push of an already-registered resource is the
-
+         * normal (re)join path, not an error */
-        if (r1 == NULL) {
+        rc = resourceApplyDef(sched->core.resmgr, r);
-            ERROR("%s add resource %s failed", __func__,
+        if (rc == RES_APPLY_CHANGED) {
+            WARN("%s resource %s definition changed in the pushed config -- "
+                 "not applied to the running scheduler (restart to pick it "
+                 "up)",
+                 __func__, r->name ? r->name : "(null)");
+        } else if (rc == RES_APPLY_SAME) {
+            DBG("%s resource %s already registered, unchanged", __func__,
                   r->name ? r->name : "(null)");
+        } else if (rc < 0) {
+            ERROR("%s add resource %s failed (rc=%d)", __func__,
+                  r->name ? r->name : "(null)", rc);
         }
     }
     return 0;

src/scheduler/testsuite/test_core_events.c

@@ -52,6 +52,7 @@
 #include <unity/unity.h>
 
 #include "kvmgr.h"
+#include "resobj.h"
 #include "scheduler.h"
 #include "xcealloc.h"
 #include "xstring.h"
@@ -1583,6 +1584,42 @@ static void test_job_rusage_reconciles_mem_reservation_against_usage(void)
  * xce_conf4sched_t payload). Also asserts on_queue_update fires once per
  * embedded queue (design intent: a config push's queues behave the same as
  * individual queue updates, just batched). */
+/* bug-1854: every scheduler (re)join re-pushes CONFIG, and each pushed
+ * resource was re-added with makeResource(), which refuses an existing name
+ * -> one ERROR per resource per rejoin (33 in data/xce_scheduler.log), and
+ * a genuinely changed definition was dropped just as silently. A re-push of
+ * the same definition is a no-op; a changed one is reported, not applied. */
+static void test_resource_reapply_same_is_noop_changed_is_reported(void)
+{
+    xce_sched_core_t core;
+    xce_resource_t   r;
+    xce_resource_t*  reg = NULL;
+    int              rc;
+
+    fixture_init(&core, &g_full_plugin);
+
+    memset(&r, 0, sizeof(r));
+    r.name    = "bug1854res";
+    r.valtype = XCE_VAL_NUMERIC;
+    r.ordtype = XCE_ORD_INCR;
+    r.flags   = XCE_RESF_CONSUMABLE;
+
+    TEST_ASSERT_EQUAL_INT(RES_APPLY_ADDED, resourceApplyDef(core.resmgr, &r));
+    TEST_ASSERT_EQUAL_INT_MESSAGE(RES_APPLY_SAME,
+        resourceApplyDef(core.resmgr, &r),
+        "re-pushing an identical definition must be a quiet no-op");
+
+    r.flags = 0; /* the definition changed */
+    TEST_ASSERT_EQUAL_INT(RES_APPLY_CHANGED, resourceApplyDef(core.resmgr, &r));
+
+    lock(core.resmgr, RTTI_LOCK_READ);
+    rc = key_find(core.resmgr, (void*)"bug1854res", (void**)&reg);
+    unlock(core.resmgr);
+    TEST_ASSERT_TRUE(rc > 0);
+    TEST_ASSERT_EQUAL_INT_MESSAGE(XCE_RESF_CONSUMABLE, reg->flags,
+        "a changed definition is reported, not applied");
+}
+
 static void test_config_push_decomposes_queues_resources_and_resvals(void)
 {
     xce_sched_core_t   core;
@@ -1899,6 +1936,7 @@ int main(void)
     RUN_TEST(test_job_rusage_applies_state_when_hook_is_null);
     RUN_TEST(test_job_rusage_reconciles_mem_reservation_against_usage);
     RUN_TEST(test_config_push_decomposes_queues_resources_and_resvals);
+    RUN_TEST(test_resource_reapply_same_is_noop_changed_is_reported);
     RUN_TEST(test_core_build_primary_schedobj_populates_resreq_fp_from_parsed_resreq);
     RUN_TEST(test_core_build_job_wires_real_task_count_into_resreq_primary);
     RUN_TEST(test_core_build_job_default_task_count_still_one);