From 0c1bac4426ba8489a4c8bc54c0e5ad87dbd6eb06 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sun, 27 Sep 2026 03:05:20 +0000 Subject: [PATCH] fix(controller): job_ctrl lock/reference and dispatch retry requeue (bug-448, bug-1626, bug-1631) bug-448: job_ctrl() took a reference on the job but only one error path released it; every successful call leaked one (refcount 1 -> 3 after two calls). It is now released on every exit path, including the key_find rc<0 one. bug-1626: job_ctrl()'s not-started branch changes the job (mark_job_update / mark_job_fail -> apply_job_*_transition, which require RTTI_LOCK_WRITE) while holding only RTTI_LOCK_READ, so it could run concurrently with readers. It now takes the write lock. (dispatch_a_job_ already did.) bug-1631: dispatch_a_job_()'s four retryable-failure branches never reset state/allocs, correct for a first dispatch (still PEND) but not for a job that had already been dispatched once: it stayed RUN holding its old allocation. Each branch now calls record_job_reconcile_pending_fact() -- a no-op without a WAL fact for a PEND job; for a RUN job it returns it to PEND, releases allocs and appends a durable WAL_JOB_UPDATE -- and notifies after unlocking when it applied, the same pattern as the resync reconcile path (cfe66dde). Not reachable until something re-dispatches RUN jobs (e.g. preemption), fixed now so it is correct when that lands. - job_get_refcount() (jobj.c) read accessor for the leak test. - test_apply_pump.c: test_job_ctrl_releases_its_job_reference and test_job_ctrl_mutates_not_started_job_under_write_lock (both RED before). - test_dispatch_fail_retry.c now asserts PEND + allocs released after a post-success retryable failure (RED before). Controller suite 1236/0; all 111 meson suites pass. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/controller/job.c | 63 +++++++++++++- src/controller/testsuite/test_apply_pump.c | 82 +++++++++++++++++++ .../testsuite/test_dispatch_fail_retry.c | 12 ++- src/include/jobj.h | 1 + src/lib/jobj.c | 9 ++ 5 files changed, 163 insertions(+), 4 deletions(-) diff --git a/src/controller/job.c b/src/controller/job.c index 10ef92ce..2d0d10fc 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -1395,6 +1395,7 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) xce_dispatch_fail_class_t dfclass; job_failinfo_t failinfo = { .jobid = alloc->jobid }; bool resolved_retryable = false; + bool requeued = false; uint32_t failcount; lock(j, RTTI_LOCK_WRITE); @@ -1406,6 +1407,11 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) failcount = ++j->dispatch_fail_count; if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) { resolved_retryable = true; + /* bug-1631: a job that was already dispatched once must go + * back to PEND without its old allocation; a no-op (no WAL + * fact) for the usual first-dispatch case, still PEND. */ + requeued = (record_job_reconcile_pending_fact(c, j) + == (int)JOB_TRANSITION_APPLIED); } else { failinfo.code = rc; failinfo.detail = "dispatch failed: worker lookup failed " @@ -1438,8 +1444,15 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) unlock(j); + + if (requeued) + + persist_notify_job_updated_already_persisted(c, j); + + dereference(j); + if (resolved_retryable) { notify_dispatch(c, ep, alloc->jobid, rc); } @@ -1452,6 +1465,7 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) xce_dispatch_fail_class_t dfclass; job_failinfo_t failinfo = { .jobid = alloc->jobid }; bool resolved_retryable = false; + bool requeued = false; uint32_t failcount; /* scheduler-dispatch-commit task 3.3: capture errno FIRST, before @@ -1473,6 +1487,11 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) failcount = ++j->dispatch_fail_count; if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) { resolved_retryable = true; + /* bug-1631: a job that was already dispatched once must go + * back to PEND without its old allocation; a no-op (no WAL + * fact) for the usual first-dispatch case, still PEND. */ + requeued = (record_job_reconcile_pending_fact(c, j) + == (int)JOB_TRANSITION_APPLIED); } else { failinfo.code = dfrc; failinfo.detail = "dispatch failed: worker not found " @@ -1500,8 +1519,15 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) unlock(j); + + if (requeued) + + persist_notify_job_updated_already_persisted(c, j); + + dereference(j); + if (resolved_retryable) { notify_dispatch(c, ep, alloc->jobid, dfrc); } @@ -1574,6 +1600,7 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) xce_dispatch_fail_class_t dfclass; job_failinfo_t failinfo = { .jobid = alloc->jobid }; bool resolved_retryable = false; + bool requeued = false; uint32_t failcount; ERROR("worker %s is not in correct state, status %x ep %p", @@ -1602,6 +1629,11 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) failcount = ++j->dispatch_fail_count; if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) { resolved_retryable = true; + /* bug-1631: a job that was already dispatched once must go + * back to PEND without its old allocation; a no-op (no WAL + * fact) for the usual first-dispatch case, still PEND. */ + requeued = (record_job_reconcile_pending_fact(c, j) + == (int)JOB_TRANSITION_APPLIED); } else { failinfo.code = dfrc; failinfo.detail = "dispatch failed: worker not ready " @@ -1630,8 +1662,15 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) unlock(j); + + if (requeued) + + persist_notify_job_updated_already_persisted(c, j); + + dereference(j); + if (resolved_retryable) { notify_dispatch(c, ep, alloc->jobid, dfrc); } @@ -1810,6 +1849,7 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) xce_dispatch_fail_class_t dfclass; job_failinfo_t failinfo = { .jobid = alloc->jobid }; bool resolved_retryable = false; + bool requeued = false; uint32_t failcount; SET_ERROR(-rc); @@ -1830,6 +1870,11 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) failcount = ++j->dispatch_fail_count; if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) { resolved_retryable = true; + /* bug-1631: a job that was already dispatched once must go + * back to PEND without its old allocation; a no-op (no WAL + * fact) for the usual first-dispatch case, still PEND. */ + requeued = (record_job_reconcile_pending_fact(c, j) + == (int)JOB_TRANSITION_APPLIED); } else { failinfo.code = rc; failinfo.detail = "dispatch failed: xce_startjob_req " @@ -1863,8 +1908,15 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) unlock(j); + + if (requeued) + + persist_notify_job_updated_already_persisted(c, j); + + dereference(j); + if (resolved_retryable) { notify_dispatch(c, ep, alloc->jobid, rc); } @@ -1901,6 +1953,8 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) unlock(c->jobmgr); if (rc < 0) { + if (j != NULL) + dereference(j); return rc; } @@ -1918,8 +1972,12 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) * -ENOENT, and every other branch sets rc itself. */ rc = 0; - /* FIXME: eliminate held multiple locks */ - lock(j, RTTI_LOCK_READ); + /* FIXME: eliminate held multiple locks + * + * bug-1626: WRITE, not READ -- the not-started branch below changes the + * job (mark_job_update / mark_job_fail -> apply_job_*_transition, which + * require the caller to hold the job under RTTI_LOCK_WRITE). */ + lock(j, RTTI_LOCK_WRITE); if (j->job.allocs != NULL) { xce_host_alloc_t** allocs = j->job.allocs->allocs; worker_t* w = NULL; @@ -2024,6 +2082,7 @@ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) } unlock(j); + dereference(j); /* bug-448: the reference taken above, on every path */ return rc; } diff --git a/src/controller/testsuite/test_apply_pump.c b/src/controller/testsuite/test_apply_pump.c index aa6d1376..0a570945 100644 --- a/src/controller/testsuite/test_apply_pump.c +++ b/src/controller/testsuite/test_apply_pump.c @@ -968,6 +968,86 @@ static void test_apply_pump_guard_does_not_break_job_ctrl_live_append(void) "PASS\n"); } +/* ── bug-448 / bug-1626: job_ctrl's own reference and lock discipline ── */ + +/* bug-448: job_ctrl takes a reference on the job and must drop it on every + * exit path; the normal (not-started) path never did. */ +static void test_job_ctrl_releases_its_job_reference(void) +{ + tap_setup(); + + job_t* j = tap_seed_job(907, JOB_STAT_PSUSP); + TEST_ASSERT_NOT_NULL(j); + unsigned long before = job_get_refcount(j); + + xce_signal_t sig = { .signo = SIGCONT, .force = 0 }; + xce_jobctrl_t ctrl = { .action = JobControl_u_siginfo, .jobid = 907 }; + ctrl.u.siginfo = &sig; + int rc = job_ctrl(&g_tap_c, &ctrl); + + sig.signo = SIGKILL; + int rc2 = job_ctrl(&g_tap_c, &ctrl); + + unsigned long after = job_get_refcount(j); + tap_teardown(); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(0, rc2); + TEST_ASSERT_EQUAL_UINT_MESSAGE(before, after, + "bug-448: job_ctrl must release the reference it takes on the job"); +} + +typedef struct { + job_t* j; + atomic_int done; + int rc; +} tap_jobctrl_arg_t; + +static void* tap_jobctrl_thread(void* arg) +{ + tap_jobctrl_arg_t* a = (tap_jobctrl_arg_t*)arg; + xce_signal_t sig = { .signo = SIGSTOP, .force = 0 }; + xce_jobctrl_t ctrl = { .action = JobControl_u_siginfo, + .jobid = a->j->job.jobid }; + ctrl.u.siginfo = &sig; + + a->rc = job_ctrl(&g_tap_c, &ctrl); + atomic_store(&a->done, 1); + return NULL; +} + +/* bug-1626: the not-started branch mutates the job (mark_job_update / + * mark_job_fail), so job_ctrl must hold the job's WRITE lock -- it may not + * proceed while another thread holds a read lock on the job. */ +static void test_job_ctrl_mutates_not_started_job_under_write_lock(void) +{ + tap_setup(); + + job_t* j = tap_seed_job(908, JOB_STAT_PEND); + TEST_ASSERT_NOT_NULL(j); + + tap_jobctrl_arg_t arg = { .j = j, .rc = -1 }; + atomic_store(&arg.done, 0); + + lock(j, RTTI_LOCK_READ); + pthread_t th; + TEST_ASSERT_EQUAL_INT(0, pthread_create(&th, NULL, tap_jobctrl_thread, + &arg)); + usleep(200 * 1000); + int done_while_reader_held = atomic_load(&arg.done); + unlock(j); + pthread_join(th, NULL); + + uint32_t state = j->job.state; + tap_teardown(); + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, done_while_reader_held, + "bug-1626: job_ctrl changed a job while another thread held its " + "read lock -- it must take RTTI_LOCK_WRITE"); + TEST_ASSERT_EQUAL_INT(0, arg.rc); + TEST_ASSERT_TRUE(state & JOB_STAT_PSUSP); +} + /* ── 4. Merge-staleness regression: FAIL must not clobber a racing FINISH ─ * * Reproduces the round-1 corruption bug found by review: a FINISH fact @@ -1769,6 +1849,8 @@ void test_apply_pump(void) RUN_TEST(test_apply_pump_run_engages_live_fence_on_failed_lsn); RUN_TEST(test_apply_pump_run_self_clears_fence_on_successful_retry); RUN_TEST(test_apply_pump_guard_does_not_break_job_ctrl_live_append); + RUN_TEST(test_job_ctrl_releases_its_job_reference); + RUN_TEST(test_job_ctrl_mutates_not_started_job_under_write_lock); RUN_TEST(test_apply_pump_fail_racing_finish_merges_against_current_state); RUN_TEST(test_apply_pump_late_pending_apply_registration_self_heals); RUN_TEST(test_register_pending_apply_race_against_concurrent_apply_self_heals); diff --git a/src/controller/testsuite/test_dispatch_fail_retry.c b/src/controller/testsuite/test_dispatch_fail_retry.c index 810d4924..b718395f 100644 --- a/src/controller/testsuite/test_dispatch_fail_retry.c +++ b/src/controller/testsuite/test_dispatch_fail_retry.c @@ -1058,8 +1058,16 @@ static void test_rpc_failure_bound_resets_after_intervening_success(void) "Req 4.3: the counter must count purely from the post-success " "reset, not from the job's total pre-success failure history"); - /* NOT asserted here: job.state == JOB_STAT_PEND / job.allocs == - * NULL after this retryable failure. A kiro-debug investigation + /* bug-1631 (fixed): a retryable failure of a job that had already + * been dispatched returns it to PEND and releases its allocation, + * durably (record_job_reconcile_pending_fact). */ + TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j->job.state, + "bug-1631: a retryable dispatch failure must return the job to " + "PEND even after a prior successful dispatch"); + TEST_ASSERT_NULL_MESSAGE(j->job.allocs, + "bug-1631: the job's prior allocation must be released"); + + /* History: originally NOT asserted. A kiro-debug investigation * (HIGH confidence; bug-1631) found this exact sub-scenario -- a * retryable dispatch failure occurring while the job's * controller-side state is already RUN, from step 2's real diff --git a/src/include/jobj.h b/src/include/jobj.h index 452bbf82..72b8737e 100644 --- a/src/include/jobj.h +++ b/src/include/jobj.h @@ -9,6 +9,7 @@ extern job_submit_data_t* makeJobsub(job_submit_data_t*); extern job_startinfo_t* makeJobstartinfo(job_startinfo_t*); extern job_t* makeJob(var, xce_job_t*); +extern unsigned long job_get_refcount(job_t*); extern int job_obj_init(int, int); extern void job_obj_fini(); diff --git a/src/lib/jobj.c b/src/lib/jobj.c index dd2c3a3e..15c7be27 100644 --- a/src/lib/jobj.c +++ b/src/lib/jobj.c @@ -708,6 +708,15 @@ static void job_dereference(meta_obj_t* obj) return; } +/* Current reference count of `j` (diagnostics / leak tests, bug-448). */ +unsigned long job_get_refcount(job_t* j) +{ + meta_obj_t* obj = RTTI_DATA_TO_OBJ(j); + job_private_t* priv = (job_private_t*)RTTI_OBJ_TO_PRIV(obj); + + return (unsigned long)atomic_load(&(priv->refcount)); +} + static int job_lock(meta_obj_t* obj, uint64_t option) { job_private_t* priv = (job_private_t*)RTTI_OBJ_TO_PRIV(obj);