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

feat(raft-apply-decoupling): defer job construction past the commit barrier (task 3.1)

Browse source 
Reorder handle_jobsubmit() so job objects are constructed via
create_job() only after the WAL commit succeeds, not before. Adds
jobid_range_exists() for a fast-fail pre-check with no WAL work, and
persist_submit_job_payloads_and_commit() (persist.c) operating on raw
xce_job_t* payloads instead of pre-existing job_t*. Commit remains
synchronous -- converting to the continuation-based commit API is
task 3.2.

Fixes bug-1408 (found during review): the post-commit array-
construction loop suppressed notification for an entire array batch
if any one sibling element failed to construct, stranding other
durably-committed and successfully-constructed jobs invisible until
the next restart's replay. Notification now fires unconditionally
for every successfully-constructed element, independent of sibling
failures; the client reply still separately reports the batch's
construction status.
This commit is contained in:
Rongsong Shen 2026-07-04 04:44:43 +00:00
parent 8febd663d5
commit 6f1fa97fab
6 changed files with 871 additions and 399 deletions
Download patch file Download diff file Expand all files Collapse all files

260
src/controller/job.c
View file

@ -50,36 +50,76 @@ static job_t* create_job(cluster_t* c, xce_job_t* j)
return job;
}
uint64_t job_submit(void* c, xce_job_submit_t* submit)
/* True if any jobid in [base_jobid, base_jobid + array_size) already exists
* in c->jobmgr (raft-apply-decoupling task 3.1). Checked BEFORE any WAL work
* so a colliding submission is still rejected immediately with no side
* effect, matching the fast-fail behavior create_job()'s own key_find
* guard used to provide when job construction ran before the commit
* barrier — now that construction is deferred to after commit, a collision
* detected only post-commit could no longer be cleanly reported as "never
* happened". */
static bool jobid_range_exists(cluster_t* c, uint64_t base_jobid,
uint64_t array_size)
{
job_t* job = NULL;
uint64_t base_id = submit->jobid;
xce_job_t j = { .submit = submit->submit };
time_t now = time(NULL);
int i;
for (uint64_t i = 0; i < array_size; i++) {
job_t* existing = NULL;
int found;
j.sub_time = now;
j.queue_time = now;
j.start_time = 0;
j.finish_time = 0;
j.pidinfo = NULL;
j.state = JOB_STAT_PEND;
j.allocs = NULL;
j.run_time = 0;
j.run_usage = NULL;
j.exit_info = NULL;
j.ckptinfo = NULL;
j.pending_ctrl = 0;
lock(c->jobmgr, RTTI_LOCK_READ);
found = key_find(c->jobmgr, (void*)(base_jobid + i),
(void**)&existing);
unlock(c->jobmgr);
for (i = 0; i < submit->array_size; i++) {
j.jobid = base_id + i;
job = create_job(c, &j);
if (job == NULL) {
return -errno;
if (found > 0 && existing != NULL)
return true;
}
return false;
}
/* Build one xce_job_t submission payload per array element, WITHOUT
* constructing or registering any job object (raft-apply-decoupling task
* 3.1 / Req 2.1, 2.3) — these are plain heap structs carrying only the data
* needed to build each entry's WAL payload; `free_job_submit_payloads` frees
* them. `submit->submit` is borrowed (owned by the caller's request), never
* duplicated or freed here. Returns 0 on success, -ENOMEM on allocation
* failure (payloads already stored are left for the caller to free). */
static int build_job_submit_payloads(uint64_t base_jobid,
xce_job_submit_t* submit, xlist_t* plist)
{
time_t now = time(NULL);
xlist_init(plist, submit->array_size);
for (uint64_t i = 0; i < submit->array_size; i++) {
xce_job_t* jp = calloc(1, sizeof(*jp));
if (jp == NULL)
return -ENOMEM;
if (xlist_store(plist, jp) < 0) {
free(jp);
return -ENOMEM;
}
jp->submit = submit->submit;
jp->jobid = base_jobid + i;
jp->sub_time = now;
jp->queue_time = now;
jp->state = JOB_STAT_PEND;
}
return base_id;
return 0;
}
static int xlist_free_job_payload_(xce_job_t* j, void* arg)
{
(void)arg;
free(j);
return 0;
}
static void free_job_submit_payloads(xlist_t* plist)
{
xlist_iter(plist, (xlist_iter_fn_t)xlist_free_job_payload_, NULL);
xlist_fini(plist);
}
int load_jobs(void* c)
@ -562,8 +602,10 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
uint64_t jobid;
xce_job_submit_t* submit;
xce_reply_t reply;
xlist_t jlist;
job_t* j = NULL;
xlist_t plist; /* xce_job_t* WAL payloads, pre-commit */
xlist_t jlist; /* job_t* for the post-commit notify hook */
uint64_t* lsns = NULL;
job_t* j = NULL;
int i;
int rc;
@ -574,15 +616,7 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
if (NULLSTR(submit->submit->user) || NULLSTR(submit->submit->queue)
|| (submit->submit->start_info == NULL)
|| NULLSTR(submit->submit->start_info->cmdline)) {
jobid = 0;
errno = EINVAL;
} else {
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_NEW, c, submit);
jobid = job_submit(c, submit);
}
if (jobid == 0) {
reply.status = -errno;
reply.data_which = RequestReply_data_nodata;
@ -590,9 +624,114 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
return -errno;
}
/* Build job list for the AFTER_EVENT hook BEFORE replying to client.
* The AFTER_EVENT hook writes WAL and waits for majority replication —
* the client must not be acknowledged until the job is durable. */
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT,
HOOK_DETAIL_JOB_NEW, c, submit);
jobid = submit->jobid;
if (jobid_range_exists(c, jobid, submit->array_size)) {
errno = EEXIST;
reply.status = -errno;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return -errno;
}
/* Build only the WAL payload data for each array element — no job
* object is constructed or inserted into c->jobmgr until the commit
* below succeeds (raft-apply-decoupling task 3.1 / Req 2.1, 2.3): a
* commit that fails or times out must leave the job table untouched,
* and the only way to guarantee that structurally is to never
* construct anything until commit has already returned success. */
rc = build_job_submit_payloads(jobid, submit, &plist);
if (rc == 0) {
lsns = calloc((size_t)plist.count, sizeof(*lsns));
if (lsns == NULL)
rc = -ENOMEM;
}
if (rc < 0) {
free_job_submit_payloads(&plist);
reply.status = rc;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return rc;
}
/* Explicit raft step (raft-apply-decoupling item #1 / bug-1306; task
* 3.1): WAL-append + majority-commit barrier for every payload, reading
* the commit verdict DIRECTLY. This used to run inside the AFTER_EVENT
* hook (persist_job_hook_fun), but xce_run_hooks returns only the LAST
* hook's rc, so a later notify subscriber (sched/apiserver) returning 0
* CLOBBERED the -ETIMEDOUT — the client was acked for an entry that
* never reached majority, which a new-term leader then legally
* discarded (committed-job-loss). The client must not be acknowledged,
* and no job object may exist, until the submission is durable on a
* majority. */
int commit_rc = persist_submit_job_payloads_and_commit(&c->persist,
&plist, lsns);
if (commit_rc < 0) {
WARN("handle_jobsubmit: job not durable (rc=%d) — rejecting submission",
commit_rc);
free_job_submit_payloads(&plist);
free(lsns);
reply.status = commit_rc;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return commit_rc;
}
/* Committed on a majority — NOW construct and register each job object
* (moved here from before the commit barrier; raft-apply-decoupling
* task 3.1 / Req 2.1, 2.2, 2.3). A construction failure at this point
* cannot be rolled back (the entry is already durably committed); it is
* a genuine apply-time failure, not a retryable one (design.md Error
* Strategy) — logged loud, and the submission is NOT reported as a
* success. */
int construct_rc = 0;
for (i = 0; i < plist.count; i++) {
xce_job_t* jp = (xce_job_t*)xlist_get(&plist, i);
job_t* job = create_job(c, jp);
if (job == NULL) {
int saved_errno = errno;
ERROR("handle_jobsubmit: job %lu committed at lsn %lu but "
"could not be constructed (errno=%d)",
(unsigned long)jp->jobid, (unsigned long)lsns[i],
saved_errno);
if (construct_rc == 0)
construct_rc = -saved_errno;
continue;
}
job->submit_lsn = lsns[i];
}
free_job_submit_payloads(&plist);
free(lsns);
/* Build the job list for the AFTER_EVENT hook — the same lookup used
* before this task, just run after construction (which now itself runs
* after commit) instead of before either.
*
* IMPORTANT (raft-apply-decoupling task 3.1 remediation): this lookup
* and the hook it feeds run UNCONDITIONALLY here, even when
* construct_rc < 0 for one or more array elements. Every element whose
* create_job() succeeded above is already durably committed AND already
* inserted into c->jobmgr — key_find below naturally finds only those
* (a failed element was never inserted, so it is silently absent from
* jlist with no extra bookkeeping needed). Both the scheduler and the
* apiserver are purely event-driven off this hook with no periodic
* reconciliation, so returning early here (as a previous version of
* this function did whenever ANY sibling failed to construct) would
* strand the successfully constructed siblings — real, durable jobs —
* invisible until the next controller restart replays the WAL. A
* whole-batch construction failure must never suppress notification for
* the elements that individually succeeded. */
xlist_init(&jlist, submit->array_size);
for (i = 0; i < submit->array_size; i++) {
@ -615,31 +754,13 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
}
}
/* Explicit raft step (raft-apply-decoupling item #1 / bug-1306): WAL-append
* + majority-commit barrier, reading the commit verdict DIRECTLY. This used
* to run inside the AFTER_EVENT hook (persist_job_hook_fun), but xce_run_hooks
* returns only the LAST hook's rc, so a later notify subscriber (sched/
* apiserver) returning 0 CLOBBERED the -ETIMEDOUT — the client was acked for
* an entry that never reached majority, which a new-term leader then legally
* discarded (committed-job-loss). The client must not be acknowledged until
* the job is durable on a majority. */
int commit_rc = persist_submit_wal_and_commit(&c->persist, &jlist);
if (commit_rc < 0) {
WARN("handle_jobsubmit: job not durable (rc=%d) — rejecting submission",
commit_rc);
xlist_iter(&jlist, (xlist_iter_fn_t)xlist_deref_job_item_, NULL);
xlist_fini(&jlist);
reply.status = commit_rc;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return commit_rc;
}
/* Committed on a majority — fire the AFTER_EVENT hook as NOTIFICATION only
* (scheduler push, apiserver SSE), observing already-committed state. Its
* return no longer gates durability: a notify-hook failure cannot un-commit
* an already-durable job, so it is logged but never fails the client reply. */
/* Committed + constructed — fire the AFTER_EVENT hook as NOTIFICATION
* only (scheduler push, apiserver SSE), observing already-committed,
* already-applied state. Its return no longer gates durability: a
* notify-hook failure cannot un-commit an already-durable job, so it is
* logged but never fails the client reply. Fired for every
* successfully constructed element regardless of construct_rc (see the
* comment above jlist's construction). */
int notify_rc = xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB,
HOOK_AFTER_EVENT,
HOOK_DETAIL_JOB_NEW, c, &jlist);
@ -651,7 +772,20 @@ int handle_jobsubmit(cluster_t* c, endpoint_t* ep, xce_request_t* request)
xlist_fini(&jlist);
/* Reply success only AFTER WAL write + majority commit succeeded. */
if (construct_rc < 0) {
/* At least one array element durably committed but could not be
* constructed (see the ERROR logged at the point of failure above).
* The client learns the submission did not fully succeed; the
* siblings that DID construct successfully were already notified
* above and must never have that notification gated on this. */
reply.status = construct_rc;
reply.data_which = RequestReply_data_nodata;
xce_reply(ep, c->myhost, &reply);
return construct_rc;
}
/* Reply success only AFTER WAL write + majority commit + construction. */
reply.status = 0;
reply.data_which = RequestReply_data_jobid;
reply.data.jobid = jobid;