wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

feat(raft-apply-decoupling): split fact receipt to durable-append-only (task 4.1)

Browse source 
Convert on_job_start/on_job_finish/on_job_fail/on_job_postdone to
validate and durably append only -- BEFORE_EVENT validation hook
unchanged, AFTER_EVENT (post-mutation) hook no longer fires at
receipt, live state mutation removed. Adds persist_submit_job_fact()
(persist.c) since the durable append previously lived inside the
very AFTER_EVENT hook this task silences; wire format is unchanged
and identical to what apply_committed_entry() already routes.

mark_job_start/mark_job_finish (single caller each) renamed in place
to record_job_start_fact/record_job_finish_fact. mark_job_fail/
mark_job_update (shared with job_ctrl's client-command path, out of
boundary) left untouched; new durable-append-only siblings
record_job_fail_fact/record_job_update_fact added instead.

Accepts a narrow, permanent regression: on_job_postdone's finish_time
anomaly backstop has no wire-format equivalent and is dropped
(affects only a rare no-prior-exit-status path).

Two critical hazards discovered by review, documented in tasks.md,
MUST be resolved before task 6 wires apply_committed_entry() into a
live path: (1) persist_job_hook_fun still fires on HOOK_DETAIL_JOB_UPDATE
and will double-append every live-applied fact; (2) record_job_fail_fact/
record_job_update_fact precompute a final state that can be silently
clobbered by a concurrent intervening fact once apply is deferred by
the commit round-trip.
This commit is contained in:
Rongsong Shen 2026-07-04 07:00:09 +00:00
parent f357215a6d
commit 23cbfb48e1
5 changed files with 492 additions and 116 deletions
Download patch file Download diff file Expand all files Collapse all files

188
src/controller/job.c
View file

@ -14,8 +14,10 @@
#include "xce/types.h"
#include "xstring.h"
static int mark_job_start(cluster_t*, job_t*, job_pidinfo_t*);
static int mark_job_finish(cluster_t*, job_t*, xce_jobfinish_t*);
static int record_job_start_fact(cluster_t*, job_t*, job_pidinfo_t*);
static int record_job_finish_fact(cluster_t*, job_t*, xce_jobfinish_t*);
static int record_job_fail_fact(cluster_t*, job_t*, job_failinfo_t*);
static int record_job_update_fact(cluster_t*, job_t*, int);
static int mark_job_fail(cluster_t*, job_t*, job_failinfo_t*);
static int mark_job_update(cluster_t*, job_t*, int);
@ -875,12 +877,14 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
}
/* 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. */
* mark_job_start (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2; mark_job_start
* itself was renamed record_job_start_fact by task 4.1, which stopped calling
* this core from the receipt path — see that function's own doc comment).
* 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)
{
@ -917,14 +921,27 @@ job_transition_result_t apply_job_start_transition(job_t* job,
return JOB_TRANSITION_APPLIED;
}
static int mark_job_start(cluster_t* c, job_t* job, job_pidinfo_t* info)
/* raft-apply-decoupling task 4.1 (Req 5.1): renamed from mark_job_start —
* this is now durable-append-only. The pre-mutation BEFORE_EVENT validation
* hook still fires at the identical point, unchanged; the state mutation
* (apply_job_start_transition) and the AFTER_EVENT notification hook both
* relocate to run from apply_committed_entry() (task 2.2) once this fact's
* WAL entry commits to a majority — driven by the future apply pump
* (task 6). Until then, a worker-reported START has no live effect on job
* state; only the durable append happens here. */
static int record_job_start_fact(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. */
* precedes the (now-deferred) 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. Calling apply_job_start_transition here on the real
* `job` is safe and mutates nothing — its own internal guard re-check
* hits the identical condition and returns before any assignment; the
* call exists only to produce its WARN log line (logs + no-op,
* unchanged from before this task). */
if (job_state_is_backward(job->job.state, JOB_STAT_RUN)) {
apply_job_start_transition(job, info); /* logs + no-op, unchanged */
return 0;
@ -933,10 +950,13 @@ static int mark_job_start(cluster_t* c, job_t* job, job_pidinfo_t* info)
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);
/* Durable append only — no mutation, no AFTER_EVENT firing here (Req
* 5.1, 5.3). JOB_STAT_RUN is the resultant state a START always
* produces once past the guard above; apply_committed_entry() will
* re-validate the same guard against whatever state is current AT
* APPLY TIME before setting it. */
(void)persist_submit_job_fact(&c->persist, WAL_JOB_START, job->job.jobid,
JOB_STAT_RUN);
return 0;
}
@ -961,7 +981,7 @@ int on_job_start(cluster_t* c, job_pidinfo_t* info)
}
lock(job, RTTI_LOCK_WRITE);
mark_job_start(c, job, info);
record_job_start_fact(c, job, info);
unlock(job);
dereference(job);
@ -970,11 +990,13 @@ int on_job_start(cluster_t* c, job_pidinfo_t* info)
}
/* 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
* mark_job_finish (raft-apply-decoupling task 2.1, Req 3.3/7.1/7.2; mark_job_finish
* itself was renamed record_job_finish_fact by task 4.1, which stopped calling
* this core from the receipt path — see that function's own doc comment).
* 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)
@ -1014,14 +1036,18 @@ job_transition_result_t apply_job_finish_transition(job_t* job,
return JOB_TRANSITION_APPLIED;
}
static int mark_job_finish(cluster_t* c, job_t* job, xce_jobfinish_t* finish)
/* raft-apply-decoupling task 4.1 (Req 5.1): renamed from mark_job_finish —
* durable-append-only, mirroring record_job_start_fact above. */
static int record_job_finish_fact(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. */
* precedes the (now-deferred) 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;
@ -1030,10 +1056,12 @@ static int mark_job_finish(cluster_t* c, job_t* job, xce_jobfinish_t* finish)
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);
/* Durable append only — no mutation, no AFTER_EVENT firing here (Req
* 5.1, 5.3). The resultant state is deterministic from finish->exit_code
* alone, independent of current job state. */
(void)persist_submit_job_fact(&c->persist, WAL_JOB_FINISH, job->job.jobid,
(finish->exit_code == 0) ? JOB_STAT_DONE
: JOB_STAT_EXIT);
return 0;
}
@ -1058,7 +1086,7 @@ int on_job_finish(cluster_t* c, xce_jobfinish_t* finish)
}
lock(job, RTTI_LOCK_WRITE);
mark_job_finish(c, job, finish);
record_job_finish_fact(c, job, finish);
unlock(job);
dereference(job);
@ -1100,6 +1128,12 @@ job_transition_result_t apply_job_fail_transition(job_t* job,
return JOB_TRANSITION_APPLIED;
}
/* Used only by job_ctrl's SIGKILL/SIGTERM not-started-job branch (a client
* command path, do_jobcontrol_op — out of raft-apply-decoupling task 4.1's
* boundary) — mutate-then-notify, unchanged. record_job_fail_fact below is
* the separate, durable-append-only sibling used by the worker-fact receipt
* path (on_job_fail); the two must stay independent since job_ctrl's
* synchronous command semantics are not this task's concern. */
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,
@ -1113,6 +1147,43 @@ static int mark_job_fail(cluster_t* c, job_t* job, job_failinfo_t* failinfo)
return 0;
}
/* raft-apply-decoupling task 4.1 (Req 5.1): the worker-fact receipt sibling
* of mark_job_fail above — durable-append-only, used only by on_job_fail.
* mark_job_fail itself is untouched (job_ctrl still needs its mutate-then-
* notify semantics; that command path is out of this task's boundary).
*
* mark_job_fail has no monotonic-rank guard (a FAIL always applies), so
* unlike record_job_start_fact/record_job_finish_fact there is no
* peek/rejected branch here — BEFORE_EVENT fires unconditionally, exactly as
* mark_job_fail's always does today.
*
* The resultant state depends on the CURRENT job state (PERR merges onto an
* existing DONE/EXIT terminal state rather than resetting it — job.c's
* pre-existing apply_job_fail_transition switch). That computation is
* reproduced here on a throwaway shallow copy of `job` — apply_job_fail_
* transition only ever touches `job->job.state`/`job->job.finish_time`
* (scalar fields, never pidinfo/exit_info), so a shallow copy is safe to
* discard afterward with no cleanup — rather than duplicating the switch,
* so this can never drift from what apply time will eventually compute. */
static int record_job_fail_fact(cluster_t* c, job_t* job,
job_failinfo_t* failinfo)
{
job_t scratch;
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_FAIL, c, failinfo, job);
scratch = *job;
apply_job_fail_transition(&scratch, failinfo);
/* Durable append only — no mutation of the real `job`, no AFTER_EVENT
* firing here (Req 5.1, 5.3). */
(void)persist_submit_job_fact(&c->persist, WAL_JOB_FAIL, job->job.jobid,
(int)scratch.job.state);
return 0;
}
/* 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
@ -1148,6 +1219,12 @@ job_transition_result_t apply_job_update_transition(job_t* job, int state)
return JOB_TRANSITION_APPLIED;
}
/* Used only by job_ctrl's SIGCONT/SIGSTOP suspend-resume branches (a client
* command path, do_jobcontrol_op — out of raft-apply-decoupling task 4.1's
* boundary) — mutate-then-notify, unchanged. record_job_update_fact below is
* the separate, durable-append-only sibling used by the worker-fact receipt
* path (on_job_postdone); the two must stay independent since job_ctrl's
* synchronous command semantics are not this task's concern. */
static int mark_job_update(cluster_t* c, job_t* job, int state)
{
int cur = job->job.state;
@ -1175,6 +1252,37 @@ static int mark_job_update(cluster_t* c, job_t* job, int state)
return 0;
}
/* raft-apply-decoupling task 4.1 (Req 5.1): the worker-fact receipt sibling
* of mark_job_update above — durable-append-only, used only by
* on_job_postdone. mark_job_update itself is untouched (job_ctrl still needs
* its mutate-then-notify semantics; that command path is out of this task's
* boundary). `state` here is the caller's already-computed target state
* (on_job_postdone computes it from the job's current state before calling
* this, exactly as it always has — that computation is untouched by this
* task; only what happens with the computed value changes). */
static int record_job_update_fact(cluster_t* c, job_t* job, int state)
{
int cur = job->job.state;
/* Peek the identical guards apply_job_update_transition() re-checks
* internally — same no-observable-behavior-change rationale as
* mark_job_update's peek above. */
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);
/* Durable append only — no mutation, no AFTER_EVENT firing here (Req
* 5.1, 5.3). */
(void)persist_submit_job_fact(&c->persist, WAL_JOB_UPDATE, job->job.jobid,
state);
return 0;
}
int on_job_fail(cluster_t* c, job_failinfo_t* failinfo)
{
job_t* job = NULL;
@ -1195,7 +1303,7 @@ int on_job_fail(cluster_t* c, job_failinfo_t* failinfo)
}
lock(job, RTTI_LOCK_WRITE);
mark_job_fail(c, job, failinfo);
record_job_fail_fact(c, job, failinfo);
unlock(job);
dereference(job);
@ -1298,12 +1406,22 @@ int on_job_postdone(cluster_t* c, endpoint_t* ep, uint64_t jobid)
"has not been set",
job->job.jobid);
state |= JOB_STAT_EXIT;
/* ensure that job finish time has been set */
job->job.finish_time = time(NULL);
/* raft-apply-decoupling task 4.1 (Req 5.1): the backstop
* `job->job.finish_time = time(NULL)` write that used to live here
* is removed — no state mutation may happen at receipt time.
* finish_time is not part of the wal_jobstate_t wire format (only
* {jobid, state} travels — a pre-existing limitation, see
* apply_committed_entry's doc comment in persist.h), so this
* defensive backstop has no equivalent at apply time either; a
* post-done arriving with no prior exit status recorded no longer
* gets a defensive finish_time set anywhere. Low-severity residual,
* flagged in this task's status report — not fixed here (would
* require wire-format enrichment, explicitly out of this task's
* scope per persist.h's Apply Seam doc comment). */
}
state |= JOB_STAT_PDONE;
mark_job_update(c, job, state);
record_job_update_fact(c, job, state);
unlock(job);
dereference(job);