fix(scheduler): re-PEND only on real dispatch failures, not the success ack

The controller acknowledges EVERY dispatched job through
NotifyData_data_jobDispatchFail (controller/job.c handle_jobdispatch):
a negative code is a real rejection, code 0 is the success ack (the else
branch calls notify_dispatch(jobid, 0)). sched_handle_notify treated any
jobDispatchFail as a failure and returned the job to PEND, so every
successfully dispatched job was immediately re-PENDed, re-elected, and
re-dispatched while the original instance kept running — spawning
duplicate instances cluster-wide. On a 3-worker cluster 24 submitted jobs
ran 38 instances and piled onto the first worker (t1) because its slots
were counted for each duplicate.

Fix: only sched_requeue_pend_job() when fail->code < 0; consume the
success ack (code 0) as a no-op — the job is already RUN
(sched_commit_decision marks it at commit). This was a regression of the
bug-1494 re-PEND logic (bug-1510 fix) that lacked the code guard.

Verified live on t1/t2/t3: 24x sleep-300 jobs distributed exactly 8/8/8,
zero duplicates, all DONE. Add white-box test
test_handle_dispatch_fail_requeues_only_on_negative_code.

Co-Authored-By: Claude <noreply@anthropic.com>

src/scheduler/src/handler.c

@@ -528,11 +528,17 @@ int sched_handle_notify(xce_scheduler_t* sched, xce_notify_data_t* notify)
             notify->data.sched_update_queue = NULL;
         break;
     case NotifyData_data_jobDispatchFail: {
-        /* The controller rejected a dispatch the scheduler committed (e.g.
+        /* The controller acknowledges every dispatched job through this
-         * the chosen worker went UNAVAIL between election and dispatch).
+         * event — a NEGATIVE code is a real dispatch rejection (e.g. the
-         * Return the job to PEND so it is re-elected next session — without
+         * chosen worker went UNAVAIL between election and dispatch), a
-         * this the job stays RUN in the scheduler's view and is never
+         * code of 0 is the success ack (controller/job.c handle_jobdispatch
-         * dispatched again (job stuck PEND cluster-wide). */
+         * calls notify_dispatch(code=0) on the else branch). Only a real
+         * failure must return the job to PEND for re-election — re-PENDing
+         * on the success ack returns every dispatched job to the PEND set,
+         * so the next session re-elects and re-dispatches it while the
+         * original instance keeps running, spawning duplicate instances
+         * (multi-worker over-dispatch). On success the job is already RUN
+         * (sched_commit_decision marks it at commit); just consume the ack. */
         job_failinfo_t* fail = notify->data.dispatchfail;
         char            jobidString[128];
 
@@ -541,6 +547,13 @@ int sched_handle_notify(xce_scheduler_t* sched, xce_notify_data_t* notify)
             break;
         }
 
+        if (fail->code >= 0) {
+            /* success ack — job is RUN; nothing to requeue */
+            notify->data.dispatchfail = NULL;
+            rc                        = 0;
+            break;
+        }
+
         fmt_jobid(jobidString, sizeof(jobidString), fail->jobid);
         WARN("%s: job %s dispatch failed (code %d) — returning to PEND",
              __func__, jobidString, fail->code);

src/scheduler/testsuite/test_handlers.c

@@ -30,14 +30,19 @@ int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c)
     (void)s; (void)c; return 0;
 }
 
+/* Test-controlled job for the jobDispatchFail handler case (below). */
+static xce_sched_job_t* test_dispatch_job = NULL;
+
 xce_sched_job_t* sched_get_job(xce_scheduler_t* sched, uint64_t jobid)
 {
-    (void)sched; (void)jobid; return NULL;
+    (void)sched; (void)jobid;
+    return test_dispatch_job;
 }
 
 /* White-box stub for sched_requeue_pend_job (handler.c's jobDispatchFail
- * case). sched_get_job is stubbed to NULL here, so this just returns
+ * case). The job comes from the test-controlled sched_get_job above; with
- * -ENOENT like the real function does for an unknown jobid. */
+ * test_dispatch_job set it re-PENDs it like the real function, otherwise it
+ * returns -ENOENT like the real function does for an unknown jobid. */
 int sched_requeue_pend_job(xce_scheduler_t* sched, uint64_t jobid)
 {
     (void)sched;
@@ -198,6 +203,50 @@ static void test_handle_resvals_full_skips_empty_enqueues_nonempty(void)
     free(one[0]);
 }
 
+/* ── multi-worker over-dispatch (bug-1494 class): jobDispatchFail code ── */
+
+/* The controller acknowledges EVERY dispatched job through
+ * NotifyData_data_jobDispatchFail — code 0 is the success ack
+ * (controller/job.c handle_jobdispatch's else branch), a negative code a
+ * real rejection. The scheduler must re-PEND only on a real failure;
+ * re-PENDING on the success ack returns each dispatched job to the PEND set
+ * so the next session re-dispatches it while the original instance keeps
+ * running — spawning duplicate instances across workers. */
+static void test_handle_dispatch_fail_requeues_only_on_negative_code(void)
+{
+    xce_scheduler_t   sched;
+    xce_sched_job_t   job;
+    xce_notify_data_t notify;
+    job_failinfo_t    fail;
+
+    memset(&job, 0, sizeof(job));
+    job.state = JOB_STAT_RUN; /* dispatched job is RUN in scheduler's view */
+    fixture_init(&sched);
+    test_dispatch_job = &job;
+
+    /* success ack (code 0): consumed, job stays RUN (no requeue) */
+    memset(&fail, 0, sizeof(fail));
+    fail.jobid = 42;
+    fail.code  = 0;
+    memset(&notify, 0, sizeof(notify));
+    notify.event             = NotifyData_data_jobDispatchFail;
+    notify.data.dispatchfail = &fail;
+    TEST_ASSERT_EQUAL_INT(0, sched_handle_notify(&sched, &notify));
+    TEST_ASSERT_EQUAL_INT(JOB_STAT_RUN, job.state);
+    TEST_ASSERT_NULL(notify.data.dispatchfail);
+
+    /* real failure (code < 0): returned to PEND for re-election */
+    job.state = JOB_STAT_RUN;
+    fail.code = -1;
+    memset(&notify, 0, sizeof(notify));
+    notify.event             = NotifyData_data_jobDispatchFail;
+    notify.data.dispatchfail = &fail;
+    TEST_ASSERT_EQUAL_INT(0, sched_handle_notify(&sched, &notify));
+    TEST_ASSERT_EQUAL_INT(JOB_STAT_PEND, job.state);
+
+    test_dispatch_job = NULL;
+}
+
 int main(int argc, char* argv[])
 {
     (void)argc; (void)argv;
@@ -208,5 +257,6 @@ int main(int argc, char* argv[])
     RUN_TEST(test_handle_update_resvals_skips_empty_enqueues_nonempty, 0);
     RUN_TEST(test_handle_resources_skips_empty_enqueues_nonempty, 0);
     RUN_TEST(test_handle_resvals_full_skips_empty_enqueues_nonempty, 0);
+    RUN_TEST(test_handle_dispatch_fail_requeues_only_on_negative_code, 0);
     return UNITY_END();
 }