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

fix(bug-1734): merge redelivered job state instead of discarding it

Browse source 
XCE_CORE_EVENT_JOBS_PUSH's duplicate-jobid handling (sched_add_job()
EEXIST) assumed a redelivery is always a harmless no-op -- true for
bug-1714's array-push TOCTOU race (both copies start identically
PEND), but not for a scheduler reconnect's full resync
(on_scheduler_join() -> xcesched_push_jobs(), sched.c), which
re-sends every currently-committed job through this same path
regardless of how far its state has moved on since this scheduler
last saw it.

Live-reproduced on a real 3-node HA cluster + worker: the scheduler's
outbound xcecli connection repeatedly failed over under a 50-element
array-job load; each reconnect's resync redelivered jobs that had
already finished (PEND/RUN -> DONE) while disconnected -- confirmed
via ctrl-1.log's existing bug-1731 "c->sched.ep is NULL, notify
SKIPPED" diagnostic firing for the same jobids during the gap, making
the resync that job's only remaining chance to reach the scheduler.
Discarding it unconditionally permanently froze the tracked job at
its stale state: controller shows DONE, scheduler stuck at RUN/PEND
forever.

Fix: on EEXIST, look up the existing tracked job and merge in the
redelivery's state if different (same terminal-transition
accounts/allocs cleanup XCE_CORE_EVENT_JOB_STATE_CHANGE/JOB_UPDATE
already do), instead of silently dropping it. Only state is merged;
broader field reconciliation is out of scope, same as the existing
allocs/accounts scope note this code path already carries. Bundled a
requested logging fix at the same site (raw jobid -> fmt_jobid()).

Verified live end to end: rebuilt scheduler, reran the same load
against the same cluster (reconnect storm still occurred, proving the
fix doesn't depend on suppressing it), both test arrays now show
consistent DONE state matching the controller exactly.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019QxaSzHRB7pmYFp4cVnwkA
This commit is contained in:
Rongsong Shen 2026-08-30 15:59:01 +00:00
parent 1c4a1ba24a
commit b0476a8424
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
2 changed files with 148 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

88
src/scheduler/src/core_events.c
View file

@ -743,18 +743,82 @@ int xce_core_push_events(xce_sched_core_t* core, xce_core_event_t* events,
* 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);
* ever find again.
*
* bug-1734 fix: a duplicate jobid is NOT always a harmless
* no-op redelivery. Live reproduction (50-element array
* job, real 3-node HA cluster + worker) confirmed
* on_scheduler_join()'s full resync (sched.c's
* xcesched_push_jobs(), fired on every scheduler
* reconnect -- and this scheduler's own outbound xcecli
* connection was observed failing over/reconnecting
* repeatedly under load) re-delivers EVERY currently-
* committed job through this SAME JOBS_PUSH path,
* including ones whose state has moved on since this
* scheduler last saw them (e.g. PEND/RUN -> DONE while
* disconnected -- confirmed via ctrl-1.log's
* sched_job_update_fun "c->sched.ep is NULL, notify
* SKIPPED" diagnostic firing for the same jobids during
* the reconnect gap: the resync was that job's ONLY
* remaining chance to reach this scheduler). Discarding
* the redelivery unconditionally silently and PERMANENTLY
* froze the existing tracked job at its stale state --
* exactly the "controller shows DONE, scheduler stuck at
* RUN/PEND" desync bug-1734 reported. Fix: look up the
* existing tracked job and, if the redelivery carries a
* DIFFERENT state, apply it in place (same terminal-
* transition accounts/allocs cleanup
* XCE_CORE_EVENT_JOB_STATE_CHANGE/JOB_UPDATE already do)
* instead of silently dropping it. Only `state` is
* merged -- broader field reconciliation (queue/resreq
* drift across a resync) is out of scope for this fix,
* same "documented scope decision" the existing
* job->allocs/accounts-out-of-scope note above already
* established for this code path. */
xce_sched_job_t* existing = sched_get_job(core, job->jobid);
char jobidStr[64];
/* bug-1734 (lower-priority logging request, bundled with
* the fix above since it touches this exact WARN site):
* use fmt_jobid()'s human-readable form, same as every
* other WARN in this function (see bug-1591's comment on
* fmt_jobid(), referenced at each other call site below),
* instead of a raw numeric jobid. */
fmt_jobid(jobidStr, sizeof(jobidStr), job->jobid);
WARN("%s JOBS_PUSH: sched_add_job failed for jobid %s "
"(rc=%d), merging state into existing tracked job "
"instead of discarding",
__func__, jobidStr, add_rc);
if ((existing != NULL) && (existing->state != job->state)) {
if (!JOB_IS_RUN(job->state)) {
if (existing->accounts != NULL) {
free_job_account(existing->accounts);
existing->accounts = NULL;
}
if (existing->allocs != NULL) {
free_JobAlloc_ptr(&(existing->allocs));
}
}
existing->state = job->state;
if (picker != NULL) {
picker->picker_iface->on_job_state_change(
picker, existing, job->state);
}
}
/* `job` itself was never registered in core->jobs (and
* never will be -- the existing tracked object above is
* what future sched_get_job() lookups resolve to), so it
* 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. */
FREEUP(job->resreq);
FREEUP(job->schedobj);
FREEUP(job);

72
src/scheduler/testsuite/test_core_events.c
View file

@ -527,6 +527,77 @@ test_jobs_push_duplicate_jobid_does_not_call_on_job_push_or_leak(void)
TEST_ASSERT_EQUAL_PTR(original, sched_get_job(&core, 55));
}
/* bug-1734: a redelivered JOBS_PUSH for an ALREADY-TRACKED jobid is not
* always a harmless no-op duplicate the way bug-1714's Bug A test above
* assumed. Live reproduction (50-element array job, real 3-node HA cluster +
* worker): the scheduler's own outbound xcecli connection to the controller
* repeatedly failed over and reconnected under load; every reconnect made
* the controller's on_scheduler_join() (sched.c) re-run a FULL resync --
* xcesched_push_jobs() -- which re-sends every currently-committed job
* (including ones already terminal on the controller side, e.g. DONE) via
* the SAME XceData_data_schedjobs wire path this JOBS_PUSH case handles.
* Unlike the array-push redelivery bug-1714 covers (both copies start
* identically PEND, so discarding the second is provably harmless), a
* resync redelivery can carry a job whose state has moved on since this
* scheduler last saw it (PEND/RUN -> DONE). The current code frees ANY
* duplicate unconditionally and never updates the existing tracked job, so
* that newer state is silently and PERMANENTLY lost -- confirmed live via
* controller ctrl-1.log's `c->sched.ep is NULL, notify SKIPPED` diagnostic
* (bug-1731) firing for the same jobids during the reconnect gap: the
* resync was that job's ONLY remaining chance to reach the scheduler, and
* this discard-without-merge path threw it away. Asserts a duplicate
* JOBS_PUSH carrying a DIFFERENT state for an already-tracked jobid updates
* the EXISTING xce_sched_job_t's state in place (mirroring how
* XCE_CORE_EVENT_JOB_STATE_CHANGE/JOB_UPDATE already treat a state change)
* rather than leaving it frozen at the stale value. */
static void
test_jobs_push_duplicate_jobid_with_newer_state_updates_existing_job(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(56, 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, 56);
TEST_ASSERT_NOT_NULL(tracked);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, tracked->state);
/* A full-resync redelivery (on_scheduler_join()'s xcesched_push_jobs())
* of the SAME jobid, now carrying its real/current terminal state -- the
* job finished on the controller while this scheduler's push channel
* was reconnecting. */
resynced_job = make_job(56, 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 = &resynced_job;
rc = xce_core_push_events(&core, &event, 1);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Must still resolve to the SAME xce_sched_job_t (no orphaned second
* object, no key_add() churn) ... */
TEST_ASSERT_EQUAL_PTR(tracked, sched_get_job(&core, 56));
/* ... but its state must now reflect the resync's newer information,
* not stay frozen at the stale PEND it had before the redelivery. */
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, tracked->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)
@ -1101,6 +1172,7 @@ int main(void)
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_jobs_push_duplicate_jobid_with_newer_state_updates_existing_job);
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);