diff --git a/src/scheduler/src/core_events.c b/src/scheduler/src/core_events.c index 78f1d1e2..1de9f521 100644 --- a/src/scheduler/src/core_events.c +++ b/src/scheduler/src/core_events.c @@ -1,4 +1,5 @@ #include +#include #include #include @@ -671,6 +672,34 @@ static void core_upsert_worker(xce_sched_core_t* core, xce_worker_t* w) * (WARN/ERROR) and do not abort the rest of the batch -- same discipline * scheduler.c's existing sched_update() dataq loop already follows. */ + +/* + * job_state_regresses_terminal - true if `current_state` is already a + * terminal outcome (DONE or EXIT) and `new_state` would change it to a + * DIFFERENT value. + * + * bug-1738: the controller's own terminal-transition guard + * (job_state_is_terminal(), job.c) treats DONE/EXIT as PERMANENT once set -- + * apply_job_finish_transition() no-ops a duplicate FINISH on an + * already-terminal job, and apply_job_fail_transition()'s POST_EXEC/ + * Q_POST_EXEC branch only ever MERGES a JOB_STAT_PERR flag onto whatever + * DONE/EXIT already holds, never re-decides DONE vs EXIT after the fact. + * The scheduler side had no equivalent guard: a single-job notify (in + * particular HOOK_DETAIL_JOB_FAIL's schedFailJob, which unconditionally + * requests JOB_STAT_EXIT with no regard for whether the job already reached + * DONE via an earlier, correct schedExitJob/schedUpdateJob notify) could + * silently clobber a correct terminal outcome. Shared by both single-job + * XCE_CORE_EVENT_JOB_STATE_CHANGE and the JOBS_PUSH duplicate-jobid resync + * merge (bug-1734/bug-1739) below, so both paths enforce the same + * "terminal is permanent" invariant instead of each re-deriving it. + */ +static bool job_state_regresses_terminal(unsigned int current_state, + unsigned int new_state) +{ + return (JOB_IS_DONE(current_state) || JOB_IS_EXIT(current_state)) + && (current_state != new_state); +} + int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events, int nevents) { @@ -791,7 +820,69 @@ int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events, __func__, jobidStr, add_rc); if ((existing != NULL) && (existing->state != job->state)) { - if (!JOB_IS_RUN(job->state)) { + /* bug-1738: never let a resync redelivery regress an + * already-terminal job to a DIFFERENT state -- same + * invariant XCE_CORE_EVENT_JOB_STATE_CHANGE now + * enforces above, applied here too since a full + * resync (on_scheduler_join()'s xcesched_push_jobs()) + * can just as easily carry a stale snapshot as a + * genuinely newer one. */ + if (job_state_regresses_terminal(existing->state, + job->state)) { + WARN("%s JOBS_PUSH: ignoring resync state 0x%x " + "for jobid %s already at terminal state " + "0x%x -- terminal states are permanent", + __func__, job->state, jobidStr, + existing->state); + } else if (JOB_IS_RUN(job->state)) { + /* bug-1739 (Symptom 1, "phantom RUN"): a resync + * redelivery carrying RUN must bring its OWN real + * allocation along. core_build_job() (above, at + * batch-build time) already populated + * job->allocs/job->accounts from the wire Job's + * `allocs` field -- the controller's authoritative + * dispatch record -- exactly the same way a + * fresh JOBS_PUSH for a brand-new RUN job would. + * Merging existing->state = RUN WITHOUT + * transferring this real allocation data left + * existing->state==RUN with existing->allocs== + * NULL: a job the scheduler's own resource/slot + * bookkeeping has no record of, exceeding the + * worker's real capacity in `info worker`/`info + * summary`. If the wire redelivery did NOT carry + * an allocation either (a controller that hasn't + * finished recording the dispatch yet, or an + * older/degraded payload), treat the RUN claim as + * untrustworthy and refuse the merge rather than + * manufacture a phantom -- `existing` keeps + * whatever state it already had. */ + if (job->accounts == NULL || job->allocs == NULL) { + WARN("%s JOBS_PUSH: resync claims RUN for " + "jobid %s with no allocation data -- " + "refusing to merge a phantom RUN " + "(existing state 0x%x kept)", + __func__, jobidStr, existing->state); + } else { + if (existing->accounts != NULL) { + free_job_account(existing->accounts); + } + if (existing->allocs != NULL) { + free_JobAlloc_ptr(&(existing->allocs)); + } + + existing->accounts = job->accounts; + existing->allocs = job->allocs; + job->accounts = NULL; + job->allocs = NULL; + + existing->state = job->state; + + if (picker != NULL) { + picker->picker_iface->on_job_state_change( + picker, existing, job->state); + } + } + } else { if (existing->accounts != NULL) { free_job_account(existing->accounts); existing->accounts = NULL; @@ -799,13 +890,13 @@ int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events, if (existing->allocs != NULL) { free_JobAlloc_ptr(&(existing->allocs)); } - } - existing->state = job->state; + existing->state = job->state; - if (picker != NULL) { - picker->picker_iface->on_job_state_change( - picker, existing, job->state); + if (picker != NULL) { + picker->picker_iface->on_job_state_change( + picker, existing, job->state); + } } } @@ -815,10 +906,14 @@ int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events, * must still be freed here at the same granularity * core_build_job()'s own early-failure paths use -- * job->user/queue/project are kvmgr-owned get-or-create - * lookups, not owned by this instance, so left alone; - * this redelivered clone's own allocs/accounts (as - * opposed to `existing`'s, handled above) are out of - * scope here, same as before this fix. */ + * lookups, not owned by this instance, so left alone. + * job->allocs/job->accounts were already transferred to + * `existing` above when the RUN-merge branch taken; any + * other branch leaves them as core_build_job() built + * them (out of scope for this fix, same as before) -- + * FREEUP(job) below only frees the job shell itself, not + * these two fields, matching this function's existing + * "job->allocs/accounts are out of scope" convention. */ FREEUP(job->resreq); FREEUP(job->schedobj); FREEUP(job); @@ -1004,6 +1099,31 @@ int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events, break; } + /* bug-1738: reject a single-job notify that would regress an + * already-terminal job to a DIFFERENT state. Live-confirmed + * mechanism: HOOK_DETAIL_JOB_FAIL's schedFailJob notify + * (sched_job_fail_fun, sched.c) unconditionally requests + * JOB_STAT_EXIT with no regard for the job's current state -- + * the controller's OWN apply_job_fail_transition() (job.c) only + * merges a JOB_STAT_PERR flag onto an already-DONE/EXIT job in + * this exact scenario (a post-exec-stage failure on a job whose + * real exit outcome was already decided), never re-decides + * DONE vs EXIT after the fact -- so a job that finished + * successfully (DONE) and then hit an unrelated post-exec + * script failure would otherwise have the scheduler's view + * forcibly clobbered back to EXIT here. See + * job_state_regresses_terminal()'s own doc comment above. */ + if (job_state_regresses_terminal(job->state, new_state)) { + char jobidStr[64]; + + fmt_jobid(jobidStr, sizeof(jobidStr), jobid); + WARN("%s JOB_STATE_CHANGE: ignoring state 0x%x for job %s " + "already at terminal state 0x%x -- terminal states are " + "permanent", + __func__, new_state, jobidStr, job->state); + break; + } + if (!JOB_IS_RUN(new_state)) { if (job->accounts != NULL) { free_job_account(job->accounts); diff --git a/src/scheduler/testsuite/test_core_events.c b/src/scheduler/testsuite/test_core_events.c index 9617f0fd..e3e20d7a 100644 --- a/src/scheduler/testsuite/test_core_events.c +++ b/src/scheduler/testsuite/test_core_events.c @@ -598,6 +598,237 @@ test_jobs_push_duplicate_jobid_with_newer_state_updates_existing_job(void) TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, tracked->state); } +/* bug-1739 (Symptom 1, "phantom RUN"): bug-1734's own JOBS_PUSH duplicate- + * jobid merge (test above) adopted a redelivery's RUN state without + * transferring any real allocation, leaving a tracked job with state==RUN + * but allocs==NULL -- a job the scheduler's own resource/slot bookkeeping + * has no record of at all (live-reported: `info worker` listed fewer + * "assigned job:" entries than `info jobs` showed RUN, exceeding the + * worker's real core count). core_build_job() already adopts a wire Job's + * `allocs` field into the freshly-built xce_sched_job_t (job->accounts = + * make_job_account(job->allocs)) exactly like a brand-new RUN job would -- + * this asserts a RUN-state resync redelivery that DOES carry real + * allocation data transfers it into the EXISTING tracked job instead of + * discarding it. */ +static void +test_jobs_push_duplicate_jobid_run_state_transfers_allocation(void) +{ + xce_sched_core_t core; + xce_core_event_t event; + xce_job_t* job1; + xce_job_t* resynced_job; + xce_sched_job_t* tracked; + xce_alloc_t* a_slots; + xce_host_alloc_t* h0; + xce_host_alloc_t* hallocs[1]; + xce_job_alloc_t* jalloc; + int rc; + + fixture_init(&core, &g_full_plugin); + + job1 = make_job(57, 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); + + tracked = sched_get_job(&core, 57); + TEST_ASSERT_NOT_NULL(tracked); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, tracked->state); + TEST_ASSERT_NULL(tracked->allocs); + + /* A resync redelivery carrying RUN AND the controller's real dispatch + * allocation for this job -- exactly what on_scheduler_join()'s + * xcesched_push_jobs() sends for a genuinely running job. */ + a_slots = make_alloc((char*)"slots", NULL, 0, 1.0f, 0); + h0 = make_host_alloc((char*)"hostA", 1, make_allocs(1, a_slots)); + hallocs[0] = h0; + jalloc = make_job_alloc(57, 1, make_host_allocs(1, hallocs[0])); + + resynced_job = make_job(57, JOB_STAT_RUN, "normal", "alice", + "proj1"); + resynced_job->allocs = jalloc; + + memset(&event, 0, sizeof(event)); + event.type = XCE_CORE_EVENT_JOBS_PUSH; + event.u.jobs_push.njobs = 1; + event.u.jobs_push.jobs = &resynced_job; + + rc = xce_core_push_events(&core, &event, 1); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_EQUAL_PTR(tracked, sched_get_job(&core, 57)); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, tracked->state); + /* The real allocation must have been transferred, not discarded -- + * this is the actual fix: a RUN job must never end up with allocs == + * NULL via this path. */ + TEST_ASSERT_NOT_NULL(tracked->allocs); + TEST_ASSERT_NOT_NULL(tracked->accounts); +} + +/* bug-1739 companion: if a RUN-state resync redelivery does NOT carry any + * allocation data (no `allocs` on the wire), merging it in anyway would + * manufacture exactly the same phantom-RUN phenotype from a different + * cause. Asserts the merge is refused (existing job keeps its prior state) + * rather than trusting an unallocated RUN claim. */ +static void +test_jobs_push_duplicate_jobid_run_state_without_allocation_is_refused(void) +{ + xce_sched_core_t core; + xce_core_event_t event; + xce_job_t* job1; + xce_job_t* resynced_job; + xce_sched_job_t* tracked; + int rc; + + fixture_init(&core, &g_full_plugin); + + job1 = make_job(58, 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); + + tracked = sched_get_job(&core, 58); + TEST_ASSERT_NOT_NULL(tracked); + + /* RUN claim with NO allocation data attached. */ + resynced_job = make_job(58, JOB_STAT_RUN, "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 = &resynced_job; + + rc = xce_core_push_events(&core, &event, 1); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_EQUAL_PTR(tracked, sched_get_job(&core, 58)); + /* Refused -- stays at its prior (safe) state, never becomes a phantom + * RUN with no allocation. */ + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, tracked->state); + TEST_ASSERT_NULL(tracked->allocs); +} + +/* bug-1738: the JOBS_PUSH duplicate-jobid merge must never regress an + * already-terminal job to a DIFFERENT terminal state either -- a resync can + * just as easily carry a stale snapshot as a genuinely newer one. Pushes + * jobid to DONE (as the bug-1734 test above already proves works), then a + * SECOND redelivery claiming EXIT must be ignored. */ +static void +test_jobs_push_duplicate_jobid_does_not_regress_terminal_state(void) +{ + xce_sched_core_t core; + xce_core_event_t event; + xce_job_t* job1; + xce_job_t* done_job; + xce_job_t* stale_exit_job; + xce_sched_job_t* tracked; + int rc; + + fixture_init(&core, &g_full_plugin); + + job1 = make_job(59, 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); + + done_job = make_job(59, JOB_STAT_DONE, "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 = &done_job; + + rc = xce_core_push_events(&core, &event, 1); + TEST_ASSERT_EQUAL_INT(0, rc); + + tracked = sched_get_job(&core, 59); + TEST_ASSERT_NOT_NULL(tracked); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, tracked->state); + + /* A stale/out-of-order redelivery claiming EXIT for a job already + * correctly recorded as DONE. */ + stale_exit_job = make_job(59, JOB_STAT_EXIT, "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 = &stale_exit_job; + + rc = xce_core_push_events(&core, &event, 1); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_EQUAL_PTR(tracked, sched_get_job(&core, 59)); + /* Must still be DONE -- the stale EXIT must not have clobbered it. */ + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, tracked->state); +} + +/* bug-1738 (single-job notify path): XCE_CORE_EVENT_JOB_STATE_CHANGE must + * apply the SAME "terminal is permanent" guard as the JOBS_PUSH merge above + * -- this is the path a live schedFailJob notify actually reaches + * (sched_job_fail_fun, sched.c, unconditionally requests JOB_STAT_EXIT with + * no regard for the job's current state; the controller's own + * apply_job_fail_transition() only merges JOB_STAT_PERR onto an + * already-DONE/EXIT job in the equivalent real scenario, never re-decides + * DONE vs EXIT). Pushes a job to RUN then DONE via JOB_STATE_CHANGE, then a + * second JOB_STATE_CHANGE claiming EXIT must be ignored. */ +static void test_job_state_change_does_not_regress_terminal_state(void) +{ + xce_sched_core_t core; + xce_core_event_t events[3]; + xce_job_t* job1; + xce_sched_job_t* job; + int rc; + + fixture_init(&core, &g_full_plugin); + + job1 = make_job(60, JOB_STAT_PEND, "normal", "alice", "proj1"); + + memset(events, 0, sizeof(events)); + + events[0].type = XCE_CORE_EVENT_JOBS_PUSH; + events[0].u.jobs_push.njobs = 1; + events[0].u.jobs_push.jobs = &job1; + + events[1].type = XCE_CORE_EVENT_JOB_STATE_CHANGE; + events[1].u.job_state_change.jobid = 60; + events[1].u.job_state_change.new_state = JOB_STAT_DONE; + + rc = xce_core_push_events(&core, events, 2); + TEST_ASSERT_EQUAL_INT(0, rc); + + job = sched_get_job(&core, 60); + TEST_ASSERT_NOT_NULL(job); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, job->state); + + /* A schedFailJob-shaped notify claiming EXIT for the SAME job, arriving + * after it already correctly reached DONE. */ + events[2].type = XCE_CORE_EVENT_JOB_STATE_CHANGE; + events[2].u.job_state_change.jobid = 60; + events[2].u.job_state_change.new_state = JOB_STAT_EXIT; + + rc = xce_core_push_events(&core, &events[2], 1); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Must still be DONE -- the stale/narrower EXIT notify must not have + * clobbered it. */ + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, job->state); +} + /* 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) @@ -1173,6 +1404,10 @@ int main(void) 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_jobs_push_duplicate_jobid_with_newer_state_updates_existing_job); + RUN_TEST(test_jobs_push_duplicate_jobid_run_state_transfers_allocation); + RUN_TEST(test_jobs_push_duplicate_jobid_run_state_without_allocation_is_refused); + RUN_TEST(test_jobs_push_duplicate_jobid_does_not_regress_terminal_state); + RUN_TEST(test_job_state_change_does_not_regress_terminal_state); 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);