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

feat(raft-apply-decoupling): extract guarded fact transition logic into reusable cores (task 2.1)

Browse source 
Split mark_job_start/finish/fail/update in job.c into four hook-free
core functions (apply_job_*_transition) containing only the
monotonic-rank guard and mutation. Wrappers now peek the same guard
to decide hook-firing, then delegate to the core, preserving exact
pre-extraction hook-firing timing. Prerequisite for task 2.2's
apply_committed_entry() dispatch and the ordered apply pump (6.1).
This commit is contained in:
Rongsong Shen 2026-07-03 15:26:22 +00:00
parent 2d9f1e947a
commit 41d36204d5
6 changed files with 480 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

152
src/controller/job.c
View file

@ -661,24 +661,29 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
return 0;
}
static int mark_job_start(cluster_t* c, job_t* job, job_pidinfo_t* info)
/* Guarded state-transition core for a worker-reported START, extracted from
* mark_job_start (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2). Contains
* ONLY the monotonic-rank guard and the state mutation — it fires NO hook
* itself, so hook-firing can be relocated by a later task (the shared apply
* seam) without touching this guard logic again. Safe to invoke directly,
* bypassing the worker-report receipt path (on_job_start) entirely, as long
* as the caller already holds `job` under RTTI_LOCK_WRITE. */
job_transition_result_t apply_job_start_transition(job_t* job,
job_pidinfo_t* info)
{
int cur = job->job.state;
/* Transition guard (Req 5.2): START sets the non-terminal JOB_STAT_RUN. A
* START on an already-terminal job (stale/duplicate worker report after
* the job finished) is a backward transition — reject it, preserve the
* terminal state, and fire NO hook (no WAL append). Return success-no-op
* so the caller (on_job_start) still reports success. */
* the job finished) is a backward transition — reject it and preserve
* the terminal state. */
if (job_state_is_backward(cur, JOB_STAT_RUN)) {
WARN("job %ld: rejecting backward START on terminal state 0x%x — "
"preserving terminal state",
job->job.jobid, cur);
return 0;
return JOB_TRANSITION_REJECTED_BACKWARD;
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_START, c, info, job);
/* update job information */
job->job.state = JOB_STAT_RUN;
job->job.start_time = time(NULL);
@ -696,6 +701,27 @@ static int mark_job_start(cluster_t* c, job_t* job, job_pidinfo_t* info)
job->job.pidinfo->exdata = NULL;
}
return JOB_TRANSITION_APPLIED;
}
static int mark_job_start(cluster_t* c, job_t* job, job_pidinfo_t* info)
{
/* Peek the identical guard apply_job_start_transition() re-checks
* internally, so this extraction changes NO observable behavior: today
* no hook fires on a rejected transition, and BEFORE_EVENT strictly
* precedes the mutation. The peek is a pure read of job->job.state, and
* every caller holds `job` under RTTI_LOCK_WRITE, so it cannot race with
* the core's own (redundant, defensive) re-check of the same guard. */
if (job_state_is_backward(job->job.state, JOB_STAT_RUN)) {
apply_job_start_transition(job, info); /* logs + no-op, unchanged */
return 0;
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_START, c, info, job);
apply_job_start_transition(job, info);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_START, c, info, job);
@ -730,23 +756,28 @@ int on_job_start(cluster_t* c, job_pidinfo_t* info)
return 0;
}
static int mark_job_finish(cluster_t* c, job_t* job, xce_jobfinish_t* finish)
/* Guarded state-transition core for a worker-reported FINISH, extracted from
* mark_job_finish (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2). Contains
* ONLY the monotonic-rank guard and the state mutation — fires NO hook, so
* hook-firing can be relocated later without touching this guard logic
* again. Safe to invoke directly, bypassing the worker-report receipt path
* (on_job_finish) entirely, as long as the caller already holds `job` under
* RTTI_LOCK_WRITE. */
job_transition_result_t apply_job_finish_transition(job_t* job,
xce_jobfinish_t* finish)
{
int cur = job->job.state;
/* Transition guard (Req 5.1): FINISH drives the job to a terminal state
* (DONE on exit_code 0, else EXIT). If the job is already terminal the
* FINISH has already been applied (retransmit / WAL replay) — log it and
* return a success-no-op WITHOUT firing the AFTER_EVENT hook, so no second
* WAL entry is appended and current_lsn does not advance. */
* FINISH has already been applied (retransmit / WAL replay) — treat it
* as a no-op duplicate so no second WAL entry is appended. */
if (job_state_is_terminal(cur)) {
INFO("job %ld: duplicate FINISH on terminal state 0x%x — no-op",
job->job.jobid, cur);
return 0;
return JOB_TRANSITION_NOOP_DUPLICATE;
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_EXIT, c, finish, job);
/* update job information */
if (finish->exit_code == 0) {
job->job.state = JOB_STAT_DONE;
@ -767,6 +798,27 @@ static int mark_job_finish(cluster_t* c, job_t* job, xce_jobfinish_t* finish)
job->job.run_time = job->job.finish_time - job->job.start_time;
return JOB_TRANSITION_APPLIED;
}
static int mark_job_finish(cluster_t* c, job_t* job, xce_jobfinish_t* finish)
{
/* Peek the identical guard apply_job_finish_transition() re-checks
* internally, so this extraction changes NO observable behavior: today
* no hook fires on a duplicate-terminal FINISH, and BEFORE_EVENT strictly
* precedes the mutation. The peek is a pure read of job->job.state, and
* every caller holds `job` under RTTI_LOCK_WRITE, so it cannot race with
* the core's own (redundant, defensive) re-check of the same guard. */
if (job_state_is_terminal(job->job.state)) {
apply_job_finish_transition(job, finish); /* logs + no-op, unchanged */
return 0;
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_EXIT, c, finish, job);
apply_job_finish_transition(job, finish);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_EXIT, c, finish, job);
@ -801,12 +853,19 @@ int on_job_finish(cluster_t* c, xce_jobfinish_t* finish)
return 0;
}
static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo)
/* State-transition core for a worker-reported FAIL, extracted from
* mark_job_fail (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2). Contains
* ONLY the state mutation — fires NO hook, so hook-firing can be relocated
* later without touching this logic again. mark_job_fail has no monotonic-
* rank guard today (unlike START/FINISH/UPDATE): a FAIL always applies,
* merging JOB_STAT_PERR onto an existing DONE/EXIT terminal state rather
* than being rejected — that existing merge behavior is preserved unchanged
* by this extraction, not newly added. Safe to invoke directly, bypassing
* the worker-report receipt path (on_job_fail) entirely, as long as the
* caller already holds `job` under RTTI_LOCK_WRITE. */
job_transition_result_t apply_job_fail_transition(job_t* job,
job_failinfo_t* failinfo)
{
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_FAIL, c, failinfo, job);
/* update job information */
switch (failinfo->stage) {
case JOB_EXEC_STAGE_Q_POST_EXEC:
case JOB_EXEC_STAGE_POST_EXEC:
@ -825,38 +884,77 @@ static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo)
job->job.finish_time = time(NULL);
return JOB_TRANSITION_APPLIED;
}
static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo)
{
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_FAIL, c, failinfo, job);
apply_job_fail_transition(job, failinfo);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_FAIL, c, failinfo, job);
return 0;
}
static int mark_job_update(cluster_t* c, job_t* job, int state)
/* Guarded state-transition core for an UPDATE, extracted from
* mark_job_update (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2). Contains
* ONLY the monotonic-rank guards and the state mutation — fires NO hook, so
* hook-firing can be relocated later without touching this guard logic
* again. Safe to invoke directly, bypassing the mark_job_update call sites
* entirely, as long as the caller already holds `job` under
* RTTI_LOCK_WRITE. */
job_transition_result_t apply_job_update_transition(job_t* job, int state)
{
int cur = job->job.state;
/* Transition guard (Req 5.1/5.2), sharing the backward predicate with
* START: an UPDATE that clears every terminal bit a terminal job holds is
* backward — reject it, preserve the terminal state, fire no hook. A
* forward merge that keeps a terminal bit (postdone OR'ing PDONE) is not
* backward and proceeds. A no-op update (requested state already current)
* is logged and skipped so it appends no duplicate WAL entry. */
* backward — reject it and preserve the terminal state. A forward merge
* that keeps a terminal bit (postdone OR'ing PDONE) is not backward and
* proceeds. A no-op update (requested state already current) is a
* duplicate — skip it so it appends no duplicate WAL entry. */
if (job_state_is_backward(cur, state)) {
WARN("job %ld: rejecting backward UPDATE 0x%x -> 0x%x — preserving "
"terminal state",
job->job.jobid, cur, state);
return 0;
return JOB_TRANSITION_REJECTED_BACKWARD;
}
if (state == cur) {
INFO("job %ld: duplicate UPDATE to state 0x%x — no-op", job->job.jobid,
cur);
return JOB_TRANSITION_NOOP_DUPLICATE;
}
/* update job state */
job->job.state = state;
return JOB_TRANSITION_APPLIED;
}
static int mark_job_update(cluster_t* c, job_t* job, int state)
{
int cur = job->job.state;
/* Peek the identical guards apply_job_update_transition() re-checks
* internally, so this extraction changes NO observable behavior: today
* no hook fires on a rejected/duplicate transition, and BEFORE_EVENT
* strictly precedes the mutation. The peek is a pure read of
* job->job.state, and every caller holds `job` under RTTI_LOCK_WRITE, so
* it cannot race with the core's own (redundant, defensive) re-check of
* the same guards. */
if (job_state_is_backward(cur, state) || (state == cur)) {
apply_job_update_transition(job, state); /* logs + no-op, unchanged */
return 0;
}
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, job);
/* update job state */
job->job.state = state;
apply_job_update_transition(job, state);
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_UPDATE, c, job);