fix(scheduler): don't hand an orphaned duplicate job to the picker

Found while chasing bug-1714's crash (root cause still open -- see that
entry): two related, independently-confirmed defects, not proven to be
bug-1714's own cause but real bugs on their own merits.

xce_core_push_events()'s JOBS_PUSH case never checked sched_add_job()'s
return value -- if key_add() failed (duplicate jobid), the freshly-built
xce_sched_job_t was handed to picker_iface->on_job_push() anyway, even
though it was never registered in core->jobs. An orphaned job the
picker knows about but sched_get_job() can never find again. Now checks
the return, frees the orphan (at core_build_job()'s own established
cleanup granularity -- xce_sched_job_t has no complete destructor
anywhere in this codebase, kill_sched_job() only frees ->rusages, so
this intentionally doesn't invent a more thorough one), and skips the
picker call.

This makes it safe to fix (not the other way around): sched_on_job_array()'s
per-element dedup check (handler.c) has a TOCTOU race -- a job is only
added to core->jobs later, by the driver thread, so a redelivered array
push arriving on the request thread before the driver catches up isn't
caught. Documented as a best-effort optimization, not a correctness
boundary -- the actual authoritative uniqueness check is sched_add_job()'s
key_add(), which the JOBS_PUSH fix above now handles correctly.

TDD: new test reproduces the duplicate-jobid path (RED: on_job_push
fired twice, Expected 1 Was 2), GREEN after the fix. Full scheduler
regression: 54/55, only the known bug-1561-adjacent flake.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019QxaSzHRB7pmYFp4cVnwkA

src/scheduler/src/core_events.c

@@ -723,6 +723,7 @@ int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events,
             for (j = 0; j < ev->u.jobs_push.njobs; j++) {
                 xce_sched_job_t* job
                     = core_build_job(core, ev->u.jobs_push.jobs[j]);
+                int add_rc;
 
                 if (job == NULL) {
                     WARN("%s JOBS_PUSH: failed to build job (batch index %d)",
@@ -730,7 +731,35 @@ int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events,
                     continue;
                 }
 
-                sched_add_job(core, job);
+                add_rc = sched_add_job(core, job);
+                if (add_rc < 0) {
+                    /* bug-1714 follow-up (Bug A): key_add() failed -- almost
+                     * always a duplicate jobid (sched_on_job_array()'s
+                     * per-element dedup check in handler.c is a best-effort
+                     * optimization, not an atomic guarantee -- see that
+                     * function's own comment), but this is the authoritative
+                     * uniqueness boundary regardless of cause. `job` was
+                     * NEVER registered in core->jobs, so it must not be
+                     * handed to the picker: doing so would leave the picker
+                     * tracking an orphaned xce_sched_job_t that no
+                     * core->jobs-keyed lookup (sched_get_job() included) can
+                     * ever find again. Free it instead, at the same
+                     * granularity core_build_job()'s own early-failure paths
+                     * above already use (resreq/schedobj/job itself via
+                     * FREEUP()) -- job->user/queue/project are kvmgr-owned
+                     * get-or-create lookups, not owned by this job instance,
+                     * so they are left alone; job->allocs/accounts are left
+                     * alone too (out of scope here -- see xce_sched_job_t's
+                     * documented-incomplete destructor gap, kill_sched_job()
+                     * in schedobj.c). */
+                    WARN("%s JOBS_PUSH: sched_add_job failed for jobid %llu "
+                         "(rc=%d), discarding duplicate/orphaned job",
+                         __func__, (unsigned long long)job->jobid, add_rc);
+                    FREEUP(job->resreq);
+                    FREEUP(job->schedobj);
+                    FREEUP(job);
+                    continue;
+                }
 
                 if (picker != NULL) {
                     picker->picker_iface->on_job_push(picker, job);

src/scheduler/src/handler.c

@@ -208,7 +208,33 @@ static int sched_on_job_array(xce_scheduler_t*  sched,
          * sched->jobs) the ordinary single-job path relies on downstream
          * (sched_add_job -> key_add -> -EEXIST), applied here per-element so
          * an already-materialized element is never re-queued. Not a new
-         * dedup mechanism. */
+         * dedup mechanism.
+         *
+         * bug-1714 follow-up (Bug B): this check has a TOCTOU race window,
+         * and is a best-effort optimization for the common case, NOT an
+         * atomic correctness boundary. This function runs on whatever
+         * thread handles the incoming network request; a job is only
+         * actually inserted into sched->core.jobs later, when the separate
+         * DRIVER thread drains the pending-update queue this function
+         * enqueues into (via sched_on_jobs()/enq_pend_update()) and calls
+         * xce_core_push_events() -> the JOBS_PUSH case (core_events.c) ->
+         * sched_add_job(). Between this function enqueueing a batch and the
+         * driver thread actually processing it, sched_get_job() cannot see
+         * those jobids yet -- so a redelivered (e.g. network-retried) copy
+         * of the same array push arriving on the request thread during that
+         * window is NOT caught here, and a second, independent set of
+         * xce_job_t/job_submit_data_t clones for the same jobids gets
+         * constructed and enqueued too. This is safe, not just tolerated:
+         * the redelivered duplicate still gets fully constructed and pushed
+         * through the pipeline, but is correctly and safely rejected at the
+         * ACTUAL authoritative uniqueness boundary -- sched_add_job()'s
+         * key_add() in core.c, whose failure return is now checked by
+         * core_events.c's JOBS_PUSH handler (Bug A fix) -- which frees the
+         * orphaned duplicate and never hands it to the picker. Do not treat
+         * this early check as a correctness guarantee, and do not "fix" the
+         * race here (e.g. via a new synchronized pending-jobids set) without
+         * first re-deriving why that's unnecessary now that Bug A closes the
+         * actual hazard. */
         if (sched_get_job(&(sched->core), jobid) != NULL) {
             continue;
         }

src/scheduler/testsuite/test_core_events.c

@@ -466,6 +466,67 @@ static void test_jobs_push_fires_on_job_push_once_per_job(void)
     TEST_ASSERT_NOT_NULL(sched_get_job(&core, 3));
 }
 
+/* bug-1714 follow-up (Bug A): xce_core_push_events()'s JOBS_PUSH case never
+ * checked sched_add_job()'s return value, so a redelivered duplicate jobid
+ * (sched_add_job -> key_add -> -EEXIST) was still handed to
+ * picker->picker_iface->on_job_push() even though it was never actually
+ * registered in core->jobs -- an orphaned xce_sched_job_t the picker now
+ * tracks but no core->jobs-keyed lookup (sched_get_job() included) can ever
+ * find again. Asserts: (1) baseline single push fires on_job_push exactly
+ * once and registers the job; (2) a SECOND JOBS_PUSH batch carrying the SAME
+ * jobid must NOT fire on_job_push again (spy count stays at the baseline,
+ * last-seen job pointer unchanged) and sched_get_job() for that jobid must
+ * still return the ORIGINAL job object, not the orphaned duplicate. */
+static void
+test_jobs_push_duplicate_jobid_does_not_call_on_job_push_or_leak(void)
+{
+    xce_sched_core_t core;
+    xce_core_event_t event;
+    xce_job_t*       job1;
+    xce_job_t*       dup_job;
+    xce_sched_job_t* original;
+    int              rc;
+
+    fixture_init(&core, &g_full_plugin);
+
+    job1 = make_job(55, JOB_STAT_PEND, "normal", "alice", "proj1");
+
+    memset(&event, 0, sizeof(event));
+    event.type              = XCE_CORE_EVENT_JOBS_PUSH;
+    event.u.jobs_push.njobs = 1;
+    event.u.jobs_push.jobs  = &job1;
+
+    rc = xce_core_push_events(&core, &event, 1);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+    TEST_ASSERT_EQUAL_INT(1, g_counts.job_push_calls);
+
+    original = sched_get_job(&core, 55);
+    TEST_ASSERT_NOT_NULL(original);
+    TEST_ASSERT_EQUAL_PTR(original, g_counts.last_job_push);
+
+    /* Redelivered duplicate: a second, independently-built xce_job_t with
+     * the SAME jobid, in its own JOBS_PUSH batch -- mirrors what
+     * sched_on_job_array()'s TOCTOU race window (Bug B, handler.c) can let
+     * through to this layer. */
+    dup_job = make_job(55, JOB_STAT_PEND, "normal", "alice", "proj1");
+
+    memset(&event, 0, sizeof(event));
+    event.type              = XCE_CORE_EVENT_JOBS_PUSH;
+    event.u.jobs_push.njobs = 1;
+    event.u.jobs_push.jobs  = &dup_job;
+
+    rc = xce_core_push_events(&core, &event, 1);
+    TEST_ASSERT_EQUAL_INT(0, rc);
+
+    /* on_job_push must NOT fire a second time for the duplicate. */
+    TEST_ASSERT_EQUAL_INT(1, g_counts.job_push_calls);
+    TEST_ASSERT_EQUAL_PTR(original, g_counts.last_job_push);
+
+    /* core->jobs must still resolve to the ORIGINAL job -- not an orphaned
+     * duplicate that a naive re-lookup might otherwise stumble into. */
+    TEST_ASSERT_EQUAL_PTR(original, sched_get_job(&core, 55));
+}
+
 /* JOB_UPDATE (core_events.c's live wire-shape mirror of JOB_STATE_CHANGE --
  * the notify-driven single-job path a real controller's HOOK_DETAIL_JOB_UPDATE
  * fire reaches, per handler.c's own XCE_SCHED_OP_UPDATE_JOB_FIELDS comment)
@@ -1039,6 +1100,7 @@ int main(void)
     UNITY_BEGIN();
     RUN_TEST(test_push_events_fires_each_hook_once_per_event_with_picker_idle);
     RUN_TEST(test_jobs_push_fires_on_job_push_once_per_job);
+    RUN_TEST(test_jobs_push_duplicate_jobid_does_not_call_on_job_push_or_leak);
     RUN_TEST(test_job_update_to_terminal_state_frees_accounts_and_allocs);
     RUN_TEST(test_non_picker_events_mutate_state_but_never_call_hooks);
     RUN_TEST(test_workers_push_twice_updates_existing_worker_in_place);