lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: the controller is the ground truth for job state on scheduler rejoin (bug-1860)

Browse source 
A job the controller requeued to PEND while the scheduler was disconnected
stayed RUN (with its old allocation) in the scheduler forever: the
bug-1812 guard in core_events.c's JOBS_PUSH merge refused any non-RUN,
non-terminal resync claim over a job holding a real allocation, because
it could not tell a genuine requeue from a push built before the
scheduler's own fresh dispatch.

Design (per review): the scheduler persists nothing, so on rejoin the
controller's state wins, and the race the guard protected against is
closed where the truth lives:

- scheduler JOBS_PUSH: the bug-1812 refusal is removed; the controller's
  state replaces the scheduler's and its allocation is released. The
  terminal-is-permanent and phantom-RUN (RUN without allocation data)
  guards stay.
- controller dispatch_a_job_(): refuses a job that is not PEND (RUN or
  finished) before any worker is touched -- previously nothing checked,
  so a stale second dispatch would start a running job again (or, through
  a retryable failure, requeue it). The job is left as is, no WAL fact,
  and its true state is re-sent (HOOK_DETAIL_JOB_UPDATE). A PEND job with
  a leftover allocation stays dispatchable.
- scheduler JOB_UPDATE: a RUN report carrying the controller's committed
  allocation is adopted, so a corrected job is never RUN with nothing
  charged.

Tests (all RED before): test_jobs_push_duplicate_jobid_controller_pend_
wins_over_local_run (replaces the bug-1812 test that pinned the
refusal), test_job_update_run_adopts_controller_allocation,
test_dispatch_of_non_pending_job_is_refused. Two tests that seeded a RUN
job to reach retry branches (a precondition their own headers called
unreachable) now use the real paths: the counter-reset test requeues via
record_job_reconcile_pending_fact() first; the disconnected-worker RUN
test asserts the refusal contract.

Controller suite 1238/0, scheduler 87/87. HA soaks: chaotic 6 rounds PASS
(0 divergence/loss, 6 worker-resync jobs), sequential 6 rounds PASS
(SIMVERIFY 7 files, 0 mismatches). The race did not occur in either run
(0 refusals logged), so the new paths are proven by the unit tests.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-09-27 04:32:37 +00:00
parent 0f7b467c2e
commit c69705ae54
Signed by untrusted user: rshen
GPG key ID: 80533536B0EB651C
4 changed files with 236 additions and 119 deletions
Download patch file Download diff file Expand all files Collapse all files

30
src/controller/job.c
View file

@ -1313,6 +1313,36 @@ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc)
return -errno;
}
/* bug-1860: only a PEND job may be dispatched. Nothing checked this, so
* a second dispatch for a job already RUN -- e.g. from a scheduler
* acting on a rejoin push older than the job's own dispatch -- would
* start it on a worker again (or, via a retryable failure, requeue a
* running job). Refuse it before any worker is touched, leave the job
* exactly as it is, and re-send its true state to the scheduler, which
* adopts it (the controller is the ground truth). */
{
int state;
bool dispatchable;
lock(j, RTTI_LOCK_READ);
state = j->job.state;
/* State only: a PEND job still carrying a leftover allocation
* stays dispatchable (dispatch replaces it), never refused
* forever. */
dispatchable = JOB_IS_PEND(state) && !JOB_IS_RUN(state)
&& !JOB_IS_FINISH(state);
unlock(j);
if (!dispatchable) {
WARN("dispatch_a_job_: job %s is not pending (state 0x%x) -- "
"refusing a stale dispatch and re-sending its state",
jobidStr, state);
persist_notify_job_updated_already_persisted(c, j);
dereference(j);
return -EALREADY;
}
}
nc = alloc->nhosts;
if (nc <= 0) {
/* scheduler-dispatch-commit task 3.2 (Req 4.2, 4.5): resolves this

145
src/controller/testsuite/test_dispatch_fail_retry.c
View file

@ -621,29 +621,16 @@ static void test_disconnected_worker_fails_immediately_no_retry(void)
printf("test_disconnected_worker_fails_immediately_no_retry: PASS\n");
}
/* ── worker-rejoin-resync task 2.3: JOB_TRANSITION_APPLIED wiring proof
* ────────────────────────────────────────────────────────────────────
* The NOOP_DUPLICATE test above exercises the only precondition dispatch_
* a_job_()'s real caller can currently construct (a PEND job -- the
* scheduler only ever dispatches from its PEND pool, so dispatch_a_job_()
* is never actually invoked on a RUN job today; the same "not reachable
* via any real caller in the currently-implemented system" caveat
* test_rpc_failure_bound_resets_after_intervening_success's header
* documents for the four RETRYABLE sites applies here too -- see bug-1631).
* This test instead directly seeds a RUN job with a real, committed
* job->job.allocs (mirroring test_worker_resync.c's own technique) to
* exercise record_job_reconcile_pending_fact()'s OTHER outcome
* (JOB_TRANSITION_APPLIED, not NOOP_DUPLICATE) THROUGH this exact new
* dispatch_a_job_() call site -- proving the branch's result-handling
* (unlock -> notify-only-on-APPLIED -> dereference) is wired correctly,
* complementing task 2.1's test_worker_resync.c (which already proves
* record_job_reconcile_pending_fact() itself is correct in isolation, not
* through this caller). Not a claim that THIS exact precondition (RUN job
* hitting the disconnected branch) is reachable in production today --
* design.md's own bug-1631 narrative is that record_job_reconcile_pending_
* fact() releases allocs at the point of RECONCILIATION (a separate,
* later-task call site, resync.c), so by the time any job reaches a fresh
* dispatch_a_job_() call it is already PEND with allocs already NULL. */
/* ── bug-1860: a RUN job is never re-dispatched ──────────────────────────
* Originally (worker-rejoin-resync task 2.3) this seeded a RUN job with a
* committed allocation to drive record_job_reconcile_pending_fact()'s
* APPLIED outcome through the disconnected-worker branch -- a precondition
* its own header already called unreachable from any real caller. Since
* bug-1860, dispatch_a_job_() refuses any job that is not PEND before a
* worker is touched, so it now pins that contract for the disconnected-
* worker case: refused, left RUN with its allocation, no WAL fact, state
* re-sent. (record_job_reconcile_pending_fact() itself stays covered by
* test_worker_resync.c.) */
#define DFR_JOBID_DISCONNECTED_RUN 91207ULL
static uint64_t dfr_current_lsn(void)
@ -656,7 +643,7 @@ static uint64_t dfr_current_lsn(void)
return lsn;
}
static void test_disconnected_worker_reconciles_run_job_releases_allocs(void)
static void test_disconnected_worker_run_job_redispatch_is_refused(void)
{
job_t* j;
worker_t* w = NULL;
@ -708,29 +695,24 @@ static void test_disconnected_worker_reconciles_run_job_releases_allocs(void)
dfr_arm_worker_canary(w);
rc = dispatch_a_job_test(&g_dfr, NULL, redispatch_alloc);
/* bug-1860: a job that is already RUN is refused before any worker is
* touched -- dispatching it again would start it twice. It stays
* exactly as it is (no requeue, no WAL fact), and its true state is
* re-sent to the scheduler. */
TEST_ASSERT_TRUE_MESSAGE(rc < 0,
"a dispatch to a disconnected worker must still fail");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j->job.state,
"Req 4.6/5.4: a RUN job redispatched to a disconnected worker must "
"be returned to PEND");
TEST_ASSERT_NULL_MESSAGE(j->job.allocs,
"bug-1631: the job's prior resource allocation must be released, "
"so a later redispatch never inherits a stale allocation");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(0, j->dispatch_fail_count,
"Req 5.4: dispatch_fail_count must stay untouched on this path");
"a re-dispatch of a RUN job must be refused");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state,
"bug-1860: a RUN job must not be requeued by a stale dispatch");
TEST_ASSERT_EQUAL_PTR(prior_alloc, j->job.allocs);
TEST_ASSERT_EQUAL_UINT32(0, j->dispatch_fail_count);
TEST_ASSERT_EQUAL_INT_MESSAGE(before_count, cbrp_notify_capture_count_ext(),
"notify_dispatch() (the retry-signal path) must never fire here");
lsn_after = dfr_current_lsn();
TEST_ASSERT_TRUE_MESSAGE(lsn_after > lsn_before,
"an APPLIED reconciliation must durably append a WAL_JOB_UPDATE "
"fact");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, lsn_after,
"a refused dispatch appends no WAL fact");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_dfr_update_hook_fires,
"bug-1679: an APPLIED reconciliation must notify the scheduler "
"exactly once, via this exact dispatch_a_job_() call site");
"bug-1860: the job's true state must be re-sent to the scheduler");
dfr_assert_ref_not_over_released(j, DFR_JOBID_DISCONNECTED_RUN);
dfr_assert_worker_ref_not_over_released(w, "dfr-wkr-disc2");
@ -739,7 +721,7 @@ static void test_disconnected_worker_reconciles_run_job_releases_allocs(void)
dfr_job_update_spy_hook);
free_JobAlloc_ptr(&redispatch_alloc);
dfr_teardown();
printf("test_disconnected_worker_reconciles_run_job_releases_allocs: "
printf("test_disconnected_worker_run_job_redispatch_is_refused: "
"PASS\n");
}
@ -1024,6 +1006,18 @@ static void test_rpc_failure_bound_resets_after_intervening_success(void)
dfr_drain_ep(&working_ep);
/* bug-1860: a RUN job can no longer be re-dispatched (dispatch_a_job_()
* refuses anything not PEND), so return the job to PEND the way a real
* re-admission does -- record_job_reconcile_pending_fact(), e.g. the
* worker reported it unrecognized on rejoin -- before the failures below
* exercise the post-success counter. */
lock(j, RTTI_LOCK_WRITE);
TEST_ASSERT_EQUAL_INT((int)JOB_TRANSITION_APPLIED,
record_job_reconcile_pending_fact(&g_dfr, j));
unlock(j);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state);
TEST_ASSERT_NULL(j->job.allocs);
/* Step 3: swap back to the broken endpoint and fail 3 more times -- the
* genuinely load-bearing proof (see this test's header). Each of these
* three must still resolve retryable/PEND; if the reset above had not
@ -1128,6 +1122,68 @@ static void test_rpc_failure_bound_resets_after_intervening_success(void)
"PASS\n");
}
/* bug-1860: dispatch_a_job_() never checked the job's state, so a second
* dispatch for a job already RUN (e.g. a scheduler acting on a stale rejoin
* push) would start it on a worker again. A job that is not PEND must be
* refused before any worker is touched, left exactly as it is, and its
* true state re-sent to the scheduler (HOOK_DETAIL_JOB_UPDATE). */
#define DFR_JOBID_NOT_PEND 91208ULL
static void test_dispatch_of_non_pending_job_is_refused(void)
{
job_t* j;
xce_job_alloc_t* alloc;
xce_job_alloc_t* committed;
int rc;
int before_count;
dfr_setup();
g_dfr_update_hook_fires = 0;
TEST_ASSERT_EQUAL_INT(0,
xce_add_hook(&g_dfr.hooksys, XCE_CONTROLLER_HOOK_JOB,
dfr_job_update_spy_hook, NULL));
j = dfr_seed_job(DFR_JOBID_NOT_PEND, "dfr-q-notpend");
committed = make_job_alloc(DFR_JOBID_NOT_PEND, 1,
make_host_allocs(1,
make_host_alloc((char*)"dfr-wkr-first", 1,
make_allocs(1,
make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
lock(j, RTTI_LOCK_WRITE);
j->job.state = JOB_STAT_RUN;
j->job.allocs = committed;
unlock(j);
alloc = make_job_alloc(DFR_JOBID_NOT_PEND, 1,
make_host_allocs(1,
make_host_alloc((char*)"dfr-wkr-second", 1,
make_allocs(1,
make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
TEST_ASSERT_NOT_NULL(alloc);
cbrp_notify_capture_reset_ext();
before_count = cbrp_notify_capture_count_ext();
dfr_arm_ref_canary(j);
rc = dispatch_a_job_test(&g_dfr, NULL, alloc);
TEST_ASSERT_TRUE_MESSAGE(rc < 0,
"bug-1860: dispatching a job that is already RUN must be refused");
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state);
TEST_ASSERT_EQUAL_PTR_MESSAGE(committed, j->job.allocs,
"the job's committed allocation must be left untouched");
TEST_ASSERT_EQUAL_UINT32(0, j->dispatch_fail_count);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_dfr_update_hook_fires,
"bug-1860: the job's true state must be re-sent to the scheduler");
TEST_ASSERT_EQUAL_INT_MESSAGE(before_count,
cbrp_notify_capture_count_ext(),
"a refused stale dispatch is not a retryable failure");
dfr_assert_ref_not_over_released(j, DFR_JOBID_NOT_PEND);
free_JobAlloc_ptr(&alloc);
dfr_teardown();
}
void test_dispatch_fail_retry(void)
{
UnitySetTestFile(__FILE__);
@ -1139,10 +1195,11 @@ void test_dispatch_fail_retry(void)
RUN_TEST(test_rpc_failure_resolves_retryable);
RUN_TEST(test_rpc_failure_bounded_escalation);
RUN_TEST(test_rpc_failure_bound_resets_after_intervening_success);
RUN_TEST(test_dispatch_of_non_pending_job_is_refused);
printf("\n>> dispatch_a_job_() disconnected-worker fail-fast tests "
"(worker-rejoin-resync task 2.3, Req 5.1, 5.4)\n\n");
RUN_TEST(test_disconnected_worker_fails_immediately_no_retry);
RUN_TEST(test_disconnected_worker_reconciles_run_job_releases_allocs);
RUN_TEST(test_disconnected_worker_run_job_redispatch_is_refused);
}