fix(scheduler): keep a moved job's own priority (bug-1944)

Both job-fields update paths, the core JOB_UPDATE event (core_events.c,
used by the live and sim drivers) and sched_update_existing_job()
(schedobj.c), set job->priority to the new queue's priority when a job
changed queue, while ingest gives every job 0. builtin_fcfs orders a
queue's jobs by job priority first, so a moved job overtook every job
already pending in the target queue (and snapshot restore, which sets 0,
disagreed with live for moved jobs).

Moving now only re-points job->queue; queue priority still orders the
queues themselves. test_job_move_priority.c moves a job into a priority-10
queue and checks it keeps priority 0 and does not take the only slot from
the queue's first job; test_job_update.c's in-place update case asserted
the old behaviour and now expects the job's priority to be kept.

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

src/scheduler/src/core_events.c

@@ -1363,9 +1363,12 @@ int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events,
                     xce_sched_queue_t* q
                         = sched_get_queue(core, updated->submit->queue);
 
+                    /* bug-1944: only the queue changes; job->priority is
+                     * the job's own (0 at ingest, core_build_job), so a
+                     * moved job does not overtake the target queue's jobs.
+                     * Queue priority already orders queues in the picker. */
                     if (q != NULL) {
-                        job->queue    = q;
-                        job->priority = q->priority;
+                        job->queue = q;
                     }
                 }
 

src/scheduler/src/schedobj.c

@@ -673,15 +673,15 @@ int sched_update_existing_job(xce_scheduler_t* sched, xce_job_t* updated)
 
     /* mutable submit-derived fields */
     if (updated->submit != NULL) {
-        /* queue — re-resolve the queue pointer by name. xce_job_t carries no
-         * per-job priority, so job priority follows the (possibly new) queue,
-         * consistent with the queue-derived priority model. */
+        /* queue — re-resolve the queue pointer by name. job->priority is
+         * the job's own (0 at make_sched_job) and stays as is: copying the
+         * queue's priority onto a moved job made it overtake the target
+         * queue's earlier jobs (bug-1944). */
         if (updated->submit->queue != NULL) {
             xce_sched_queue_t* q
                 = sched_get_queue(&(sched->core), updated->submit->queue);
             if (q != NULL) {
-                job->queue    = q;
-                job->priority = q->priority;
+                job->queue = q;
             }
         }
 

src/scheduler/testsuite/meson.build

@@ -3113,3 +3113,24 @@ sched_sim_group_limit_parity_test = executable(
 test('scheduler sim group limit parity test',
      sched_sim_group_limit_parity_test, suite: 'scheduler',
      env: ['ASAN_OPTIONS=detect_leaks=0'])
+
+# bug-1944: a job moved to another queue (core JOB_UPDATE event) keeps its
+# own priority and does not overtake jobs already pending in the target
+# queue. Same real-scheduler_srcs link technique as above.
+sched_job_move_priority_test_srcs = ['test_job_move_priority.c']
+foreach f : scheduler_srcs
+  if f != 'src/main.c'
+    sched_job_move_priority_test_srcs += ['../' + f]
+  endif
+endforeach
+
+sched_job_move_priority_test = executable(
+    'sched_job_move_priority_test',
+    sched_job_move_priority_test_srcs,
+    include_directories:[incdir],
+    dependencies: [scheduler_test_deps],
+    install: false)
+
+test('scheduler job move priority test',
+     sched_job_move_priority_test, suite: 'scheduler',
+     env: ['ASAN_OPTIONS=detect_leaks=0'])

src/scheduler/testsuite/test_job_move_priority.c

@@ -0,0 +1,227 @@
+/*
+ * test_job_move_priority.c - bug-1944: moving a pending job to another
+ * queue must not change the job's own priority, so the moved job does not
+ * overtake jobs submitted earlier to the target queue.
+ *
+ * Fixture: queue qa (priority 0) and queue qb (priority 10); jobs B1 and B2
+ * pending in qb (queue_time 1000, 1001), job M pending in qa (1002). M is
+ * moved to qb through the core's JOB_UPDATE event (core_events.c), the path
+ * both drivers use for a job-fields update. One worker slot, so only the
+ * first job the picker offers runs. FCFS orders a queue's jobs by job
+ * priority, then queue time: B1 must run and M must stay pending with its
+ * submit-time priority (0). Built like test_sim_contention_limit_driver_
+ * parity.c's normal-mode fixture (real scheduler_srcs, real picker and
+ * allocator threads, `static` fixture, exactly one RUN_TEST()).
+ */
+#include <errno.h>
+#include <pthread.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <unity/unity.h>
+
+#include "controller.capnp.h"
+#include "decision.h"
+#include "plugin.h"
+#include "preason.h"
+#include "resobj.h"
+#include "sched/types.h"
+#include "scheduler.h"
+#include "xstring.h"
+
+#define WORKER_NAME "jmove_worker"
+#define QUEUE_A     "jmove_qa"
+#define QUEUE_B     "jmove_qb"
+#define TEST_RESREQ ((char*)"select[type==any] rusage[mem=1]")
+
+#define JOB_B1_ID ((uint64_t)194401ULL)
+#define JOB_B2_ID ((uint64_t)194402ULL)
+#define JOB_M_ID  ((uint64_t)194403ULL)
+
+void setUp(void) { return; }
+void tearDown(void) { return; }
+
+static void register_slots_resource(var resmgr)
+{
+    xce_resource_t res;
+
+    memset(&res, 0, sizeof(res));
+    res.name    = strdup("slots");
+    res.valtype = XCE_VAL_NUMERIC;
+    res.ordtype = XCE_ORD_DECR;
+    res.flags   = XCE_RESF_LIMIT | XCE_RESF_CONSUMABLE | XCE_RESF_RELEASE
+                | XCE_RESF_DEFAULT_HOST;
+
+    TEST_ASSERT_NOT_NULL(makeResource(resmgr, &res));
+}
+
+static void add_worker(xce_sched_core_t* core, double slots)
+{
+    xce_sched_worker_t* w   = (xce_sched_worker_t*)calloc(1, sizeof(*w));
+    xce_resource_t*     r   = NULL;
+    xce_val_t*          val = NULL;
+    int                 rc;
+
+    TEST_ASSERT_NOT_NULL(w);
+    w->name = strdup(WORKER_NAME);
+    w->idx  = -1;
+    zarray_init(&(w->resvals), 128, (zarray_item_killer_t)kill_sched_resval);
+    zarray_init(&(w->resInUse), 128, (zarray_item_killer_t)kill_sched_resval);
+
+    rc = key_find(core->resmgr, (void*)"slots", (void**)&r);
+    TEST_ASSERT_TRUE(rc >= 0);
+    val = (xce_val_t*)calloc(1, sizeof(*val));
+    TEST_ASSERT_NOT_NULL(val);
+    val->which    = Value_u_fvalue;
+    val->u.fvalue = slots;
+    zarray_put(&(w->resvals), r->idx, val);
+
+    rc = sched_add_worker(core, w);
+    TEST_ASSERT_TRUE(rc >= 0);
+}
+
+static void add_queue(xce_sched_core_t* core, xce_sched_plugin_t* picker,
+                      const char* name, int priority)
+{
+    xce_sched_queue_t* q = (xce_sched_queue_t*)calloc(1, sizeof(*q));
+    int                rc;
+
+    TEST_ASSERT_NOT_NULL(q);
+    q->name     = makeString((char*)name);
+    q->priority = priority;
+    rc          = sched_add_queue(core, q);
+    TEST_ASSERT_TRUE(rc >= 0);
+    picker->picker_iface->on_queue_update(picker, q);
+}
+
+static void push_job(xce_scheduler_t* sched, xce_sched_plugin_t* picker,
+                     uint64_t jobid, const char* queue, uint64_t queue_time)
+{
+    job_submit_data_t submit;
+    xce_job_t         j;
+    xce_sched_job_t*  job;
+    int               rc;
+
+    memset(&submit, 0, sizeof(submit));
+    submit.queue        = (char*)queue;
+    submit.user         = (char*)"jmove_user";
+    submit.project_name = (char*)"default";
+    submit.resreq       = TEST_RESREQ;
+    submit.mintasks     = 1;
+    submit.maxtasks     = 1;
+
+    memset(&j, 0, sizeof(j));
+    j.jobid      = jobid;
+    j.state      = JOB_STAT_PEND;
+    j.sub_time   = queue_time;
+    j.queue_time = queue_time;
+    j.submit     = &submit;
+
+    job = make_sched_job(sched, &j);
+    TEST_ASSERT_NOT_NULL(job);
+    job->schedobj->u.primary->resreq->minTasks = 1;
+    job->schedobj->u.primary->resreq->maxTasks = 1;
+    rc = sched_add_job(&(sched->core), job);
+    TEST_ASSERT_TRUE(rc >= 0);
+    picker->picker_iface->on_job_push(picker, job);
+}
+
+static void run_round(xce_sched_core_t* core)
+{
+    pthread_t       picker_tid;
+    pthread_t       allocator_tid;
+    xce_decision_t* d;
+    int             rc;
+
+    rc = pthread_create(&picker_tid, NULL, sched_job_picker, core);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    rc = pthread_create(&allocator_tid, NULL, sched_allocator, core);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    rc = xce_core_go(core);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    do {
+        d = NULL;
+        cbuf_get(&(core->decisions), &d);
+    } while (d != NULL);
+
+    rc = xce_core_end_session(core);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+}
+
+static xce_scheduler_t g_sched;
+
+static void test_moved_job_keeps_its_priority_and_place(void)
+{
+    xce_sched_plugin_t* fcfs = NULL;
+    job_submit_data_t   move_submit;
+    xce_job_t           move;
+    xce_core_event_t    ev;
+    xce_sched_job_t*    m;
+    int                 rc;
+
+    memset(&g_sched, 0, sizeof(g_sched));
+    rc = xce_sched_core_init(&(g_sched.core));
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    register_slots_resource(g_sched.core.resmgr);
+
+    fcfs = builtin_fcfs_new_plugin();
+    TEST_ASSERT_NOT_NULL(fcfs);
+    rc = plugin_register(&g_sched, fcfs);
+    TEST_ASSERT_TRUE(rc >= 0);
+    g_sched.core.job_pick_plugin = (char*)"builtin_fcfs";
+    rc = register_seqjob_allocator(&g_sched);
+    TEST_ASSERT_TRUE(rc >= 0);
+    rc = register_parjob_allocator(&g_sched);
+    TEST_ASSERT_TRUE(rc >= 0);
+
+    add_queue(&(g_sched.core), fcfs, QUEUE_A, 0);
+    add_queue(&(g_sched.core), fcfs, QUEUE_B, 10);
+    add_worker(&(g_sched.core), 1.0);
+
+    push_job(&g_sched, fcfs, JOB_B1_ID, QUEUE_B, 1000ULL);
+    push_job(&g_sched, fcfs, JOB_B2_ID, QUEUE_B, 1001ULL);
+    push_job(&g_sched, fcfs, JOB_M_ID, QUEUE_A, 1002ULL);
+
+    m = sched_get_job(&(g_sched.core), JOB_M_ID);
+    TEST_ASSERT_NOT_NULL(m);
+    TEST_ASSERT_EQUAL_INT(0, m->priority);
+
+    /* move M from qa to qb */
+    memset(&move_submit, 0, sizeof(move_submit));
+    move_submit.queue = (char*)QUEUE_B;
+    memset(&move, 0, sizeof(move));
+    move.jobid  = JOB_M_ID;
+    move.state  = JOB_STAT_PEND;
+    move.submit = &move_submit;
+
+    memset(&ev, 0, sizeof(ev));
+    ev.type             = XCE_CORE_EVENT_JOB_UPDATE;
+    ev.u.job_update.job = &move;
+    rc = xce_core_push_events(&(g_sched.core), &ev, 1);
+    TEST_ASSERT_TRUE(rc >= 0);
+
+    TEST_ASSERT_EQUAL_STRING_MESSAGE(QUEUE_B, m->queue->name,
+                                     "the move must reach the job");
+    TEST_ASSERT_EQUAL_INT_MESSAGE(
+        0, m->priority,
+        "moving a job must not change its own priority (bug-1944)");
+
+    run_round(&(g_sched.core));
+
+    TEST_ASSERT_EQUAL_UINT_MESSAGE(
+        JOB_STAT_RUN, sched_get_job(&(g_sched.core), JOB_B1_ID)->state,
+        "B1, first in qb, must take the only slot");
+    TEST_ASSERT_EQUAL_UINT_MESSAGE(
+        JOB_STAT_PEND, m->state,
+        "the moved job must not overtake jobs already pending in qb");
+}
+
+int main(void)
+{
+    UNITY_BEGIN();
+    RUN_TEST(test_moved_job_keeps_its_priority_and_place);
+    return UNITY_END();
+}

src/scheduler/testsuite/test_job_update.c

@@ -212,7 +212,9 @@ test_update_existing_job_mutates_in_place(void)
     /* mutable fields updated */
     TEST_ASSERT_EQUAL_UINT(JOB_STAT_RUN, job->state);
     TEST_ASSERT_EQUAL_PTR(g_test_queues[0], job->queue);
-    TEST_ASSERT_EQUAL_INT(5, job->priority);
+    /* bug-1944: the job keeps its own priority; the queue's (5) is not
+     * copied onto it */
+    TEST_ASSERT_EQUAL_INT(1, job->priority);
     /* live runtime state preserved (object identity, not a fresh calloc) */
     TEST_ASSERT_EQUAL_PTR(schedobj, job->schedobj);