#include #include #include #include #include #include #include #include #include "cluster.h" #include "controller.capnp.h" #include "jobj.h" #include "persist.h" #include "preason.h" #include "request.h" #include "xce/cluster_def.h" #include "xce/errnox.h" #include "xce/types.h" #include "xstring.h" 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 mark_job_fail(cluster_t*, job_t*, job_failinfo_t*); static int mark_job_update(cluster_t*, job_t*, int); /* scheduler-dispatch-commit task 3.3: dispatch_a_job_()'s four retryable * failure sites (job.c, below) need to notify the scheduler of a retryable * failure -- notify_dispatch() is defined AFTER dispatch_a_job_() in this * file with no forward declaration until now. Mirrors mark_job_fail's own * forward declaration immediately above; body/signature unmodified. */ static void notify_dispatch(cluster_t*, endpoint_t*, uint64_t, int); /* scheduler-dispatch-commit task 3.3 (Req 4.3): default bounded-retry limit * for a job's consecutive retryable dispatch failures with no intervening * successful dispatch -- design.md's own stated default (3 consecutive * failures). The failure that pushes job.dispatch_fail_count past this * bound resolves the job to terminal instead of retrying again. */ #define XCE_DISPATCH_FAIL_RETRY_BOUND 3 /* A job is terminal once it carries a DONE or EXIT bit. Terminal jobs reject * backward (terminal -> non-terminal) transitions and treat re-applied * terminal transitions as duplicates; forward bit-OR merges (PDONE, PERR, * suspend bits) on a terminal job remain legal. */ #define JOB_STATE_TERMINAL_BITS (JOB_STAT_DONE | JOB_STAT_EXIT) static inline bool job_state_is_terminal(int state) { return (state & JOB_STATE_TERMINAL_BITS) != 0; } /* Shared backward predicate for START and UPDATE (Req 5.2): a transition is * backward when the job is already terminal but the requested next state drops * every terminal bit it currently holds (e.g. START overwriting DONE with a * bare JOB_STAT_RUN, or an UPDATE clearing the terminal bits). A forward merge * that preserves at least one terminal bit (postdone OR'ing PDONE) is NOT * backward. */ static inline bool job_state_is_backward(int cur, int next) { return job_state_is_terminal(cur) && !job_state_is_terminal(next); } static job_t* create_job(cluster_t* c, xce_job_t* j) { job_t* job = NULL; job = makeJob(c->jobmgr, j); return job; } /* bug-1419: in-flight submission registry, closing the resubmit-dedup gap * task 3.1's apply-after-commit deferral opened. * * Before task 3.1, jobid_range_exists()'s c->jobmgr check was airtight * against a client's own retried submit (xcecli_jobsubmit wrapped in * XCECLI_FAILOVER_CALL, client_failover.c) because job construction ran * synchronously in this same request: by the time a retry could possibly * reach the controller again, the first attempt's job already existed in * c->jobmgr. client_failover.c's own doc comment names this "the * controller's client-jobid replay guard" and relies on it to reclassify a * retried write's EEXIST as success (xce_failover_classify's * "Idempotent-replay rule"). * * Task 3.1 deliberately deferred job construction to AFTER commit (Req 2.1, * 2.3) — correct for its own goal (no compensating rollback needed on a * failed commit) — but this reopened exactly the window the replay guard * depended on: a submit that is durably appended and will commit, but has * not yet run through apply_committed_entry() and landed in c->jobmgr, is * invisible to jobid_range_exists(). A client retry (e.g. after a * transport error or failover mid-flight, before it ever saw the first * attempt's reply) landing in that window sails through the check and * appends a SECOND WAL_JOB_SUBMIT for the same jobid with a fresh subtime — * confirmed via waldump on a live chaotic-soak node (non-byte-identical * duplicate submits, same jobid, differing subtime/CRC), which can * ultimately trip persist_recover()'s post-replay apply-correctness check * (task 5.1) into a genuine APPLY_DIVERGENT verdict on restart. * * Fix: track jobids that have been accepted for WAL append but have not * yet had their outcome fully resolved (failed, or applied) in a small * process-local registry, and treat a hit there identically to a * c->jobmgr hit — both mean "this exact write already happened or is * already in flight," which is precisely what a retried write must see as * a duplicate. Populated in handle_jobsubmit() right after this check * passes (before any WAL work); cleared in jobsubmit_async_ctx_release() * (the single choke point already reached on every resolution path: * commit failure, apply completion, and OOM/alloc-failure early-outs). */ typedef struct inflight_submit { uint64_t base_jobid; uint64_t array_size; struct inflight_submit* next; } inflight_submit_t; static pthread_mutex_t s_inflight_submits_lock = PTHREAD_MUTEX_INITIALIZER; static inflight_submit_t* s_inflight_submits = NULL; static bool inflight_range_overlaps_locked(uint64_t base_jobid, uint64_t array_size) { for (inflight_submit_t* n = s_inflight_submits; n != NULL; n = n->next) { uint64_t lo = base_jobid, hi = base_jobid + array_size; uint64_t nlo = n->base_jobid, nhi = n->base_jobid + n->array_size; if (lo < nhi && nlo < hi) return true; } return false; } /* Atomically check-then-register under one lock hold, closing the TOCTOU a * separate check + separate add would leave between two concurrent * handle_jobsubmit calls for the same jobid (a retry's new connection can * race the still-in-flight original request under this codebase's * vtask-dispatched, not single-threaded-per-jobid, request handling). * Returns true (registered) if no overlap existed; false (already in * flight, NOT registered) if one did — the caller must reject on false * exactly as it already does for a c->jobmgr hit. */ static bool inflight_range_try_add(uint64_t base_jobid, uint64_t array_size) { bool ok = false; pthread_mutex_lock(&s_inflight_submits_lock); if (!inflight_range_overlaps_locked(base_jobid, array_size)) { inflight_submit_t* n = calloc(1, sizeof(*n)); if (n != NULL) { n->base_jobid = base_jobid; n->array_size = array_size; n->next = s_inflight_submits; s_inflight_submits = n; ok = true; } /* calloc failure: fall through with ok=false -- rejecting an * accepted-looking submit on OOM is safe (the client's own retry * logic already handles a rejected attempt); silently admitting an * untracked in-flight submission on OOM is not. */ } pthread_mutex_unlock(&s_inflight_submits_lock); return ok; } static void inflight_range_remove(uint64_t base_jobid, uint64_t array_size) { pthread_mutex_lock(&s_inflight_submits_lock); inflight_submit_t** pp = &s_inflight_submits; while (*pp != NULL) { if ((*pp)->base_jobid == base_jobid && (*pp)->array_size == array_size) { inflight_submit_t* dead = *pp; *pp = dead->next; free(dead); break; } pp = &(*pp)->next; } pthread_mutex_unlock(&s_inflight_submits_lock); } #ifdef XCE_TESTING /* Test-only seam: s_inflight_submits is process-lifetime, not tied to any * cluster_t fixture. In production that's fine -- a restart wipes it along * with every other in-memory registry, and persist_recover() rebuilds * c->jobmgr fresh from the WAL. But a unit-test binary reuses the same * process across many fixtures, and a test that deliberately leaves a * submission's commit continuation unresolved through teardown (bug-1419's * own regression test) has no other event left to release its entry -- * commit_engine_fini() frees pending continuations without invoking their * callback. Without this, that entry outlives the fixture and can spuriously * reject an unrelated later test that reuses the same jobid range. */ void inflight_submits_reset_for_test(void) { pthread_mutex_lock(&s_inflight_submits_lock); inflight_submit_t* n = s_inflight_submits; while (n != NULL) { inflight_submit_t* dead = n; n = n->next; free(dead); } s_inflight_submits = NULL; pthread_mutex_unlock(&s_inflight_submits_lock); } #endif /* 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". * * bug-1419: this check alone is no longer sufficient (see * inflight_range_try_add's doc comment) — handle_jobsubmit additionally * calls inflight_range_try_add() after this check passes, which atomically * closes the remaining not-yet-applied window this function cannot see. */ static bool jobid_range_exists(cluster_t* c, uint64_t base_jobid, uint64_t array_size) { for (uint64_t i = 0; i < array_size; i++) { job_t* existing = NULL; int found; lock(c->jobmgr, RTTI_LOCK_READ); found = key_find(c->jobmgr, (void*)(base_jobid + i), (void**)&existing); unlock(c->jobmgr); 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 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) { return 0; } /* Reconstruct a job from a WAL_JOB_SUBMIT Cap'n Proto payload. * Uses the jobid embedded in the serialized xce_job_t (not the caller's arg). * entry_lsn is the WAL entry LSN of this submit record (ha-convergence-recovery * 3.3): the restored job is tagged with it so its commit status is decidable at * scheduler-export time. 0 = unknown (pre-feature / direct create) = exportable. * Returns -EEXIST if the job is already present (idempotent replay guard). */ int restore_job(cluster_t* c, uint64_t jobid_unused, const void* capnp_payload, size_t len, uint64_t entry_lsn) { (void)jobid_unused; if (!capnp_payload || len == 0) return -EINVAL; struct capn c_ctx; capn_init_mem(&c_ctx, (const uint8_t*)capnp_payload, len, 0); Job_ptr jp; jp.p = capn_getp(capn_root(&c_ctx), 0, 1); struct Job cj; xce_job_t j; memset(&j, 0, sizeof(j)); read_Job(&cj, jp); decode_Job(&j, &cj); capn_free(&c_ctx); /* Safety guard: skip if already present (WAL replay may see duplicates * when recovery restarts from a segment boundary). */ job_t* existing = NULL; lock(c->jobmgr, RTTI_LOCK_READ); int found = key_find(c->jobmgr, (void*)j.jobid, (void**)&existing); unlock(c->jobmgr); if (found > 0 && existing != NULL) { /* j was fully decoded above (submit/allocs/pidinfo/pend_reasons all * populated by decode_Job) and create_job() is never reached on this * path, so ownership never moved -- every decoded pointer is still * ours to release here (ha-entity-snapshots task 5.2: fixes the * pre-existing decode leak on this early-return path, called out in * tasks.md's Implementation Notes). */ free_Job(&j); return -EEXIST; } job_t* job = create_job(c, &j); int saved_rc = job ? 0 : -errno; /* create_job() (jobj.c job_init) takes ownership of j.allocs/j.pidinfo/ * j.pend_reasons via MOVE_PTR (nulling them here on success) and makes * its OWN deduplicated copies of j.submit/j.ckptinfo (makeJobsub/ * makeString) -- so j.submit, the one field decode_Job always populates * that job_init never moves, is never released by create_job and must * always be freed here. free_Job() null-checks each of its four fields * before freeing (submit/allocs/pidinfo/pend_reasons), so calling it * unconditionally is safe whether create_job succeeded (three of the * four already nulled by the MOVE_PTR above) or failed before job_init * ran any MOVE_PTR (none nulled, all four correctly released) -- never a * double free either way (ha-entity-snapshots task 5.2, pre-existing * leak noted in tasks.md's Implementation Notes). */ free_Job(&j); if (!job) return saved_rc; /* ha-convergence-recovery (3.3): tag the restored job with its submit * entry's LSN so the egress watermark predicate can decide its commit * status (job_submit_committed). On a promoted primary, replayed entries * sit at or below the promotion truncation point, so restored jobs are * exportable by construction. */ job->submit_lsn = entry_lsn; return 0; } #ifdef XCE_TESTING /* job-array-single-wal-record task 3.1: forceable, per-index construction * failure for restore_job_array()'s scratch-build loop -- lets a test prove * the all-or-nothing atomicity contract (a mid-array failure discards every * element built THIS attempt, without ever touching c->jobmgr) without * needing a real OOM. -1 = disabled (default); a value >= 0 forces the * construction attempt for that 0-based element index within the CURRENT * array to fail, simulating a resource-exhaustion apply failure (Req 9.4, * 9.6). File-scope, not per-cluster_t: this task's own test only ever * drives one cluster_t fixture at a time; a later task needing independent * multi-node fault injection can revisit this. Compiled out of production * (xce_controller is built without -DXCE_TESTING). */ static _Atomic int s_job_array_apply_force_fail_at = -1; void job_array_apply_test_set_force_fail_at(int index) { atomic_store_explicit(&s_job_array_apply_force_fail_at, index, memory_order_relaxed); } void job_array_apply_test_clear_force_fail(void) { atomic_store_explicit(&s_job_array_apply_force_fail_at, -1, memory_order_relaxed); } /* job-array-single-wal-record task 3.1 remediation (review finding: the * bulk-insert pass into c->jobmgr was previously unchecked and unfailable * by test): forces the BULK-INSERT phase's key_add(c->jobmgr, ...) call -- * as opposed to the scratch-build phase's makeJob() above -- to fail at a * given 0-based index WITHIN THE BULK-INSERT PASS, simulating the one real * failure this pass can hit: c->jobmgr's backing xzarray hitting a genuine * realloc() OOM while growing under the write lock (Req 9.4). -1 = disabled * (default). Like a real allocation failure, the forced index is never * actually inserted -- indices before it were already inserted for real and * cannot be un-inserted (see the function doc comment's residual-gap * paragraph). Compiled out of production. */ static _Atomic int s_job_array_bulk_insert_force_fail_at = -1; void job_array_apply_test_set_bulk_insert_force_fail_at(int index) { atomic_store_explicit(&s_job_array_bulk_insert_force_fail_at, index, memory_order_relaxed); } void job_array_apply_test_clear_bulk_insert_force_fail(void) { atomic_store_explicit(&s_job_array_bulk_insert_force_fail_at, -1, memory_order_relaxed); } #endif /* XCE_TESTING */ /* job-array-single-wal-record task 3.1 (design.md Array Apply Expansion, * Service Interface): expands one committed WAL_JOB_ARRAY_SUBMIT payload * (base jobid, array_size, one shared job_submit_data_t) into `array_size` * job_t entries, atomically -- shared by apply_committed_entry() (live * apply / boot replay) and apply_wal_entry() (standby catch-up), task 3.2/ * 3.3, so the two dispatchers can never diverge in how they expand the same * committed array record (Req 3.1, 3.2). * * Atomicity (Req 2.1, 2.2, 2.3, 9.3, 9.4 -- scratch-build-then-bulk-insert, * NOT construct-then-rollback): every element is constructed via makeJob() * -- the same construction create_job() wraps -- against a throwaway, * function-local kv table (`scratch`, an xkvmgr_type instance identical in * kind to c->jobmgr itself, so makeJob()'s internal locking/key_find/ * key_add calls behave exactly as they do in production). c->jobmgr is * NEVER touched while this loop runs -- scratch is never visible to any * other thread, so there is no window, success path or failure path, in * which a concurrent reader of c->jobmgr (a status query, a cancel request) * can observe some array indices present and others absent. Only after * EVERY element has successfully constructed does one final pass insert * them all into c->jobmgr under a single held write-lock acquisition. Every * object here already exists, fully constructed, and no other writer can be * racing these exact jobids, so key_add() cannot fail on EEXIST or * content grounds -- but it CAN still fail on resource exhaustion: growing * c->jobmgr's own backing store (a realloc() reachable through key_add -> * xkvmgr_add -> vstore -> xzarray_store -> xzarray_ensure_items, kvmgr.c/ * zarray.c) happens WHILE this write lock is held, and c->jobmgr is sized * far below MAX_ARRAY_IDX in production (jobmgr_nhash/jobmgr_delta, * cluster.c), so a meaningfully large array was near-certain to trigger a * growth cycle here. **Task 7.1 (bug-1443) update**: this specific, * DOMINANT, N-scaling failure trigger is now closed structurally -- * kvmgr_reserve(c->jobmgr, array_size) (kvmgr.c/kvmgr.h) pre-grows * c->jobmgr's backing value store for the whole array BEFORE this loop's * first key_add() call, under this SAME write-lock acquisition. This does * NOT make key_add() unconditionally infallible, though: kvtab_add()'s own * separate, tiny per-entry hash-chain-node allocation (kv.c) and a * content-based -EEXIST collision remain theoretically possible -- both are * far rarer and non-N-scaling compared to the growth cycle just closed -- * see the residual-atomicity-gap paragraph below for how that narrower case * differs from a scratch-build-phase failure. On a construction failure at * index k during * the SCRATCH-BUILD phase (resource exhaustion only -- Req 9.1's pre-commit * validation already made a content-derived failure here impossible, Req * 9.3), the 0..k-1 scratch-built elements are destroyed directly via * job_dereference (job_t's teardown is wired through its reference * interface, not the generic destroy() -- job_type.destroy is NULL, jobj.c) * -- c->jobmgr was never mutated, so there is no key_del/rollback race to * reason about. A retry re-enters this function from index 0 against a * fresh scratch list, not from a saved offset. * * Idempotent per index (Req 3.3): if jobid already exists in c->jobmgr, * that index is skipped (not an error) -- handles re-applying an * already-fully-applied record (replay, or a resolved-but-retried commit). * If every index already exists, returns -EEXIST (mirrors restore_job()'s * own convention) and constructs nothing -- the caller must fire no hook. * * Every newly constructed element is tagged with submit_lsn = entry_lsn * (mirrors restore_job(), feeds the existing egress-watermark predicate * unchanged -- Req 5). * * out_new_jobs (optional): on a genuine (non-EEXIST) success, populated * with every newly created job_t* in index order -- structurally identical * to the ad hoc xlist_t apply_committed_entry() already builds for a single * WAL_JOB_SUBMIT's HOOK_DETAIL_JOB_NEW fire, so task 3.2 can feed it * directly to xce_run_hooks() as the array's one-shot, all-N-elements * notification (Req 3.4). Left untouched (caller must not read it) on * -EEXIST. **Task 7.1 (bug-1443) update**: on a BULK-INSERT-phase negative- * errno return (see below), out_new_jobs is now ALSO populated -- with * whatever genuinely-inserted prefix exists -- rather than left untouched; * it remains untouched only for a SCRATCH-BUILD-phase or reservation * failure, where nothing was ever inserted. * * Residual atomicity gap -- BULK-INSERT-phase failure only (narrower than * the scratch-build-phase failure above; task 7.1/bug-1443 closed the * DOMINANT trigger via kvmgr_reserve() above, but did not, and structurally * cannot, make key_add() unconditionally infallible): if key_add() itself * still fails partway through the bulk-insert pass (kvtab_add()'s own tiny * per-entry allocation, kv.c, or a content-based -EEXIST collision -- both * far rarer and non-N-scaling) (at index k of built.count), indices [0, k) * are, by that point, REAL and PERMANENTLY visible in c->jobmgr -- * xkvmgr_type exposes no key_del (kvmgr.c), so a partial bulk insert can * never be rolled back after the fact. This function still reports the * failure honestly (a negative errno, never a false built.count success) * and releases the never-inserted [k, built.count) elements so they are not * leaked, but it CANNOT restore the true "invisible-then-visible, nothing * in between" guarantee for this specific failure point. **Task 7.1 * (bug-1443) fix**: unlike before, the genuinely-inserted prefix [0, k) is * no longer silently un-notified -- it is transferred into *out_new_jobs * (when non-NULL) so the caller (apply_committed_entry()) can still fire * HOOK_DETAIL_JOB_NEW for it, even though the overall call still, * correctly, reports failure so persist_apply_pump_run() halts the LSN and * the Live Apply Resource Fence engages for the un-constructed remainder. * * Returns the count of newly created elements (1..array_size) on success, * -EEXIST if every index already existed (idempotent whole-record no-op, * no hook should fire), or a negative errno on a genuine failure. For a * reservation or SCRATCH-BUILD-phase failure, *out_new_jobs is left * untouched -- the scratch list has already been discarded and c->jobmgr * was never mutated at all (Req 9.4: no partial set ever observable). For a * BULK-INSERT-phase failure, see the residual atomicity gap paragraph above * -- c->jobmgr may already hold a real, permanent prefix of this attempt's * elements, and (task 7.1) *out_new_jobs is populated with exactly that * prefix so the caller can still notify for it. */ int restore_job_array(cluster_t* c, const void* capnp_payload, size_t len, uint64_t entry_lsn, xlist_t* out_new_jobs) { if (!c || !capnp_payload || len == 0) return -EINVAL; struct capn c_ctx; capn_init_mem(&c_ctx, (const uint8_t*)capnp_payload, len, 0); JobSubmit_ptr jp; jp.p = capn_getp(capn_root(&c_ctx), 0, 1); struct JobSubmit wire; memset(&wire, 0, sizeof(wire)); read_JobSubmit(&wire, jp); xce_job_submit_t decoded; memset(&decoded, 0, sizeof(decoded)); decode_JobSubmit(&decoded, &wire); capn_free(&c_ctx); if (decoded.submit == NULL || decoded.array_size == 0) { free_JobSubmit(&decoded); return -EINVAL; } uint64_t base_jobid = decoded.jobid; uint64_t array_size = decoded.array_size; /* Throwaway construction target -- see the function doc comment above. * datakiller is deliberately NULL: this scratch table's job_t values are * always released explicitly (via `built`, below), never through the * table's own teardown -- xkvmgr_type has no key_del, so relying on a * datakiller here would make the success path (where ownership of every * element moves to c->jobmgr) indistinguishable from the failure path * (where every element must be released) purely from the table's own * perspective. */ kvmgr_params_t scratch_params = { .key_type = KVMGR_KEY_INT, .nslots = (uint32_t)((array_size < 2) ? 2 : array_size), .delta = (uint32_t)((array_size < 1) ? 1 : array_size), .datakiller = NULL }; var scratch = makeInstance(&xkvmgr_type, &scratch_params); if (scratch == NULL) { int saved_rc = -errno; free_JobSubmit(&decoded); return saved_rc; } xlist_t built; xlist_init(&built, (int)array_size); time_t now = time(NULL); int rc = 0; for (uint64_t i = 0; i < array_size; i++) { uint64_t jobid = base_jobid + i; job_t* existing = NULL; lock(c->jobmgr, RTTI_LOCK_READ); int found = key_find(c->jobmgr, (void*)jobid, (void**)&existing); unlock(c->jobmgr); if (found > 0 && existing != NULL) continue; /* idempotent per-index skip (Req 3.3) */ xce_job_t ej; memset(&ej, 0, sizeof(ej)); ej.submit = decoded.submit; /* shared/borrowed across every * element -- job_init's makeJobsub() * interns/refcounts it, never takes * ownership; freed once, below. */ ej.jobid = jobid; ej.sub_time = now; ej.queue_time = now; ej.state = JOB_STAT_PEND; #ifdef XCE_TESTING int forced = atomic_load_explicit(&s_job_array_apply_force_fail_at, memory_order_relaxed); if (forced >= 0 && (uint64_t)forced == i) { errno = ENOMEM; /* simulated resource-exhaustion failure */ rc = -ENOMEM; break; } #endif job_t* job = makeJob(scratch, &ej); if (job == NULL) { rc = -errno; break; } job->submit_lsn = entry_lsn; if (xlist_store(&built, job) < 0) { dereference(job); /* never touched c->jobmgr -- safe to release * directly (job_type.destroy is NULL; the * reference interface is the only teardown * path, jobj.c). */ rc = -ENOMEM; break; } } /* The one shared spec is always released here, exactly once, whether * every element succeeded or not -- mirrors restore_job()'s * unconditional free_Job() discipline. */ free_JobSubmit(&decoded); if (rc < 0) { /* All-or-nothing: discard everything built THIS attempt. c->jobmgr * was never mutated, so there is nothing to roll back there -- a * retry starts over from index 0 against a fresh scratch list. */ for (int k = 0; k < built.count; k++) { job_t* j = (job_t*)xlist_get(&built, k); if (j != NULL) dereference(j); } xlist_fini(&built); destroy(scratch); return rc; } if (built.count == 0) { /* Every index already existed -- idempotent whole-record no-op * (Req 3.3). Nothing was constructed; *out_new_jobs is left * untouched and no hook should fire. */ xlist_fini(&built); destroy(scratch); return -EEXIST; } /* One bulk pass, held under a single write-lock acquisition so no * concurrent reader of c->jobmgr can ever observe a partial insert -- * this is what makes the array atomically invisible-then-visible on the * SUCCESS path too, not only recoverable on the failure path (Req 2.3, * 9.4). Every object already exists, fully constructed, and no other * writer can be racing these exact jobids (only this apply path ever * constructs them), so key_add() cannot fail on EEXIST or content * grounds -- but it CAN still fail on resource exhaustion (c->jobmgr's * backing store growing via realloc() under this write lock; see the * function doc comment's "Residual atomicity gap" paragraph). Checked * and reported below, never silently claimed as full success. */ lock(c->jobmgr, RTTI_LOCK_WRITE); /* job-array-single-wal-record task 7.1 (bug-1443): pre-grow c->jobmgr's * backing value store for `array_size` additional entries BEFORE the * first key_add() below can run -- this structurally closes the * DOMINANT, N-scaling failure trigger for this pass (the large * realloc() growing c->jobmgr's value store). array_size (rather than * the possibly-smaller built.count) is a deliberate, always-sufficient * upper bound: built.count <= array_size always (idempotent per-index * skips above only ever shrink it), so this never under-reserves, and * over-reserving by a few slots in the rare partial-skip case is * harmless. * * The `+ 1` headroom is deliberate, not padding: xzarray_ensure_items() * (zarray.c, unmodified) treats capacity as sufficient only when * `mincap < caps` (a STRICT less-than) -- verified directly by tracing * its logic -- so reserving EXACTLY `array_size` still leaves the very * LAST key_add() of a full `array_size`-element bulk-insert pass * hitting `mincap == caps`, which that strict check treats as * insufficient and grows again (by a small, delta-bounded amount, not * an N-scaling one, but still an avoidable realloc() under this write * lock). Reserving one extra slot closes that off-by-one so the * bulk-insert loop below never calls into growth at all when * reservation succeeds. * * Held under the SAME write-lock acquisition as the bulk-insert loop * itself -- not a redundant, separate lock/unlock cycle * (kvmgr_reserve()'s contract, kvmgr.h: caller must hold the kvmgr write * lock for the full duration of the call). This does NOT make * key_add() unconditionally infallible -- see the function doc * comment's "Residual atomicity gap" paragraph for what remains * theoretically possible after this call succeeds. */ int reserve_rc = kvmgr_reserve(c->jobmgr, (int)array_size + 1); if (reserve_rc < 0) { unlock(c->jobmgr); /* Reservation failure is a pre-mutation failure, exactly like a * scratch-build-phase failure above: no key_add() ever ran, so * c->jobmgr's key->index table was never touched -- there is * nothing to roll back there. Discard every scratch-built element * from THIS attempt and report the failure honestly (Req 9.4). */ for (int k = 0; k < built.count; k++) { job_t* j = (job_t*)xlist_get(&built, k); if (j != NULL) dereference(j); } xlist_fini(&built); destroy(scratch); return reserve_rc; } int inserted = 0; int insert_rc = 0; for (int k = 0; k < built.count; k++) { job_t* j = (job_t*)xlist_get(&built, k); int add_rc; #ifdef XCE_TESTING int forced_bulk = atomic_load_explicit( &s_job_array_bulk_insert_force_fail_at, memory_order_relaxed); if (forced_bulk >= 0 && forced_bulk == k) { errno = ENOMEM; /* simulated resource-exhaustion failure */ add_rc = -ENOMEM; } else #endif { add_rc = key_add(c->jobmgr, (void*)j->job.jobid, (void*)j); } if (add_rc < 0) { insert_rc = add_rc; break; } inserted++; } unlock(c->jobmgr); if (insert_rc < 0) { /* See the function doc comment's "Residual atomicity gap" paragraph: * indices [0, inserted) are already real and permanently visible in * c->jobmgr (no key_del exists to undo them) -- only the * never-inserted [inserted, built.count) tail is still ours to * release here. */ for (int k = inserted; k < built.count; k++) { job_t* j = (job_t*)xlist_get(&built, k); if (j != NULL) dereference(j); } /* job-array-single-wal-record task 7.1 (bug-1443 partial-prefix * fix): indices [0, inserted) are genuinely, permanently in * c->jobmgr -- they must never be silently un-notified. Transfer * exactly that prefix into *out_new_jobs (when the caller wants it) * so apply_committed_entry() can still fire HOOK_DETAIL_JOB_NEW for * it, even though this call still, correctly, reports failure via a * negative rc so the caller halts/fences for the un-constructed * remainder (Req 2.3, 9.4, 9.6). Truncating built.count to * `inserted` before the transfer is what excludes the already- * released [inserted, built.count) tail above from the transferred * list -- xlist_fini() only ever frees the list's own backing * array, never walks its elements, so this truncation is the only * step needed to make the transfer safe. */ if (out_new_jobs != NULL && inserted > 0) { built.count = inserted; *out_new_jobs = built; } else { xlist_fini(&built); } destroy(scratch); return insert_rc; } int created = built.count; if (out_new_jobs != NULL) *out_new_jobs = built; /* transfer ownership of built's data array */ else xlist_fini(&built); destroy(scratch); return created; } /* Apply a recorded job state transition during WAL recovery. * Looks up the job (which must already exist from WAL_JOB_SUBMIT replay) * and overwrites its state with the value stored in the WAL entry. */ int restore_job_state(cluster_t* c, uint64_t jobid, uint32_t state) { job_t* j = NULL; lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&j); unlock(c->jobmgr); if (!j) return -ENOENT; lock(j, RTTI_LOCK_WRITE); j->job.state = (int)state; unlock(j); return 0; } /* bug-1678: WAL_JOB_DISPATCH's own sibling of restore_job_state() above, for * apply_wal_entry()'s standby catch-up / replicated-apply seam -- takes * ownership of an optional decoded resource allocation (NULL for a * state-only WAL_JOB_DISPATCH entry, e.g. one written before this fix) so a * promoted standby recovers EXEC_HOST the same way apply_committed_entry's * WAL_JOB_DISPATCH case already does for the primary's own boot replay. * * bug-1697: also takes a wire-carried `start_time` (0 meaning "not carried * -- leave job.start_time untouched", matching every sibling restore_job_* * function's own "0 == not carried" convention). job.start_time is now * DISPATCH-owned (see wal_types.h's WAL_JOB_START doc comment for the full * rationale), so this is the seam that must restore it on a promoted * standby, not WAL_JOB_START's own restore_job_start_state() below. */ int restore_job_dispatch_state(cluster_t* c, uint64_t jobid, uint32_t state, uint64_t start_time, xce_job_alloc_t* alloc) { job_t* j = NULL; lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&j); unlock(c->jobmgr); if (!j) { if (alloc != NULL) free_JobAlloc_ptr(&alloc); return -ENOENT; } lock(j, RTTI_LOCK_WRITE); j->job.state = (int)state; if (start_time != 0) j->job.start_time = (time_t)start_time; if (alloc != NULL) { if (j->job.allocs != NULL) free_JobAlloc_ptr(&(j->job.allocs)); j->job.allocs = alloc; } unlock(j); return 0; } /* bug-1687: WAL_JOB_START's own sibling of restore_job_dispatch_state() * above (bug-1678), for apply_wal_entry()'s standby catch-up / replicated- * apply seam -- takes ownership of an optional decoded job_pidinfo_t (NULL * for a state-only WAL_JOB_START entry, e.g. one written before this fix) * so a promoted standby recovers the worker-confirmed pids/pgids/exdata the * same way apply_committed_entry's WAL_JOB_START case already does for the * primary's own boot replay. * * bug-1697: no longer takes/touches start_time at all -- job.start_time is * now exclusively DISPATCH-owned (restore_job_dispatch_state() above), so * this function must leave it alone entirely rather than risk clobbering * the correct dispatch-time value with a stale worker-ack-time one (see * wal_types.h's WAL_JOB_START doc comment for the full rationale). */ int restore_job_start_state(cluster_t* c, uint64_t jobid, uint32_t state, job_pidinfo_t* pidinfo) { job_t* j = NULL; lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&j); unlock(c->jobmgr); if (!j) { if (pidinfo != NULL) free_JobPIDInfo_ptr(&pidinfo); return -ENOENT; } lock(j, RTTI_LOCK_WRITE); j->job.state = (int)state; if (pidinfo != NULL) { if (j->job.pidinfo != NULL) free_JobPIDInfo_ptr(&(j->job.pidinfo)); j->job.pidinfo = pidinfo; } unlock(j); return 0; } /* bug-1689: WAL_JOB_FINISH's own sibling of restore_job_start_state()/ * restore_job_dispatch_state() above (bug-1687/bug-1678), for * apply_wal_entry()'s standby catch-up / replicated-apply seam -- takes * ownership of an optional decoded job_exitinfo_t (NULL for a state-only * WAL_JOB_FINISH entry, e.g. one written before this fix) and a * wire-carried finish_time (0 meaning "not carried -- leave * job.finish_time/run_time untouched", matching this seam's own pre-fix * behavior of never touching them for WAL_JOB_FINISH at all) so a promoted * standby recovers the worker-confirmed exit_code and the true finish_time * the same way apply_committed_entry's WAL_JOB_FINISH case already does for * the primary's own boot replay. job_exitinfo_t has no nested pointers (a * single int), so it is released with plain FREEUP, not a capnp * free_*_ptr helper. */ int restore_job_finish_state(cluster_t* c, uint64_t jobid, uint32_t state, uint64_t finish_time, job_exitinfo_t* exit_info) { job_t* j = NULL; lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&j); unlock(c->jobmgr); if (!j) { FREEUP(exit_info); return -ENOENT; } lock(j, RTTI_LOCK_WRITE); j->job.state = (int)state; if (finish_time != 0) j->job.finish_time = (time_t)finish_time; if (exit_info != NULL) { FREEUP(j->job.exit_info); j->job.exit_info = exit_info; } if (finish_time != 0) j->job.run_time = (int)(j->job.finish_time - j->job.start_time); unlock(j); return 0; } /* bug-1690: merge (never wholesale free-and-replace) a freshly-decoded * xce_job_runusage_t into *dst. Takes ownership of `src` -- always * consumed/freed by this call (either adopted whole, partially cannibalized * for its new-host entries, or fully released), regardless of outcome. * * WHY A MERGE, NOT A REPLACE: *dst can be the job's LIVE run_usage -- * update_job_rusages() (job.c) mutates it synchronously and in place, via * update_host_runusage() (misc.c), BEFORE the corresponding WAL fact is even * appended. On the primary, persist_apply_pump_run() (persist.c) later * re-applies that SAME fact via apply_committed_entry() moments afterward, * on the SAME live job object -- so *dst can genuinely be that live, * already-current structure at the moment this runs, not just a stale * replay artifact. Critically, update_host_runusage() sources each * xce_host_runusage_t.host_name from makeString() (src/lib/xstring.c) -- an * RTTI-managed, refcounted intern-table pointer, NOT a plain malloc() * allocation. RTTI's own header-before-data layout (librtti's * RTTI_DATA_TO_OBJ: the object header sits BEFORE the pointer callers hold) * means a plain free() on that pointer -- which the capnp-generated * free_JobRunusage_ptr()/free_HostRunusage_ptr() would do, being written for * genuinely capnp-decoded, plain-malloc'd Text fields -- corrupts the heap. * (job_destroy()'s own manual FREEUP-based walk over run_usage, jobj.c, has * this exact same latent mismatch already, independent of this fix -- flagged * separately, not fixed here, since jobj.c/misc.c are outside this fix's * scope.) * * The merge below NEVER frees an existing host_name: an already-present * host (matched by name) only has its scalar cpu_time/mem_usage overwritten * in place; only a genuinely NEW host's own freshly-decoded (always * plain-malloc'd) host_name is adopted, by ownership transfer, never a * live/xstring one. `src`'s own shell plus every entry NOT transferred is * released via the ordinary capnp free path at the end -- safe, since `src` * is always capnp-decoded, end to end, regardless of what *dst was. */ void merge_job_runusage(xce_job_runusage_t** dst, xce_job_runusage_t* src) { xce_job_runusage_t* d; int i, k; if (*dst == NULL) { /* Nothing live to preserve — adopt the decoded copy outright, * exactly like every other restore_job_*_state() sibling. */ *dst = src; return; } d = *dst; /* Scalar job-level fields: no pointers, always safe to overwrite in * place regardless of *dst's provenance. */ d->max_mem = src->max_mem; d->avg_mem = src->avg_mem; d->used_mem = src->used_mem; d->cpu_time = src->cpu_time; d->peak_cpu_time = src->peak_cpu_time; d->peak_cpu_duration = src->peak_cpu_duration; d->mem_efficiency = src->mem_efficiency; d->avg_cpu_efficiency = src->avg_cpu_efficiency; d->peak_cpu_efficiency = src->peak_cpu_efficiency; for (i = 0; i < src->num_hosts; i++) { xce_host_runusage_t* sh = src->host_runusage[i]; int matched = -1; if (sh == NULL) continue; for (k = 0; k < d->num_hosts; k++) { if (d->host_runusage[k] != NULL && d->host_runusage[k]->host_name != NULL && sh->host_name != NULL && strcmp(sh->host_name, d->host_runusage[k]->host_name) == 0) { matched = k; break; } } if (matched >= 0) { /* Existing host: overwrite scalars only. d's host_name pointer * (possibly xstring-backed) is never touched or freed here. */ d->host_runusage[matched]->cpu_time = sh->cpu_time; d->host_runusage[matched]->mem_usage = sh->mem_usage; d->host_runusage[matched]->mem_swap = sh->mem_swap; d->host_runusage[matched]->tasks_current = sh->tasks_current; } else { /* Genuinely new host: append, transferring ownership of sh's * own plain-malloc'd host_name directly -- never xstring. */ xce_host_runusage_t** grown = (xce_host_runusage_t**)realloc( d->host_runusage, (size_t)(d->num_hosts + 1) * sizeof(xce_host_runusage_t*)); if (grown == NULL) continue; /* OOM -- skip this host, keep the rest merged */ d->host_runusage = grown; d->host_runusage[d->num_hosts] = sh; d->num_hosts++; src->host_runusage[i] = NULL; /* ownership transferred -- the * free_JobRunusage_ptr() call * below must not also free it */ } } /* Release src's shell + every entry whose ownership was NOT transferred * above -- always safe, src is capnp-decoded end to end. */ free_JobRunusage_ptr(&src); } /* bug-1690: WAL_JOB_RUSAGE's own sibling of restore_job_state() above, for * apply_wal_entry()'s standby catch-up / replicated-apply seam -- takes * ownership of an optional decoded xce_job_runusage_t (NULL for a * state-only WAL_JOB_RUSAGE entry -- either one written before this fix, or * one fired while job.run_usage was still NULL at write time) so a promoted * standby recovers the client-reported per-host CPU/mem usage the same way * apply_committed_entry's WAL_JOB_RUSAGE case already does for the primary's * own boot replay. Unconditional absolute state overwrite, matching every * other restore_job_*_state() sibling on this seam -- no transition-result * guarding happens on standby catch-up. Merges (never frees-and-replaces) * run_usage -- see merge_job_runusage()'s doc comment above for why. */ int restore_job_rusage_state(cluster_t* c, uint64_t jobid, uint32_t state, xce_job_runusage_t* runusage) { job_t* j = NULL; lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&j); unlock(c->jobmgr); if (!j) { if (runusage != NULL) free_JobRunusage_ptr(&runusage); return -ENOENT; } lock(j, RTTI_LOCK_WRITE); j->job.state = (int)state; if (runusage != NULL) merge_job_runusage(&(j->job.run_usage), runusage); unlock(j); return 0; } /* bug-1691: WAL_JOB_CKPT's own sibling of restore_job_state() above, for * apply_wal_entry()'s standby catch-up / replicated-apply seam -- takes * ownership of an optional decoded ckptinfo string (NULL for a state-only * WAL_JOB_CKPT entry, e.g. one written before this fix, or one fired while * job.ckptinfo was still NULL at write time) so a promoted standby recovers * the checkpoint info the same way apply_committed_entry's WAL_JOB_CKPT * branch already restores it on the primary's own boot replay. Absolute * state overwrite, matching every other restore_job_*_state() sibling on * this seam -- no transition-result guarding happens on standby catch-up. * * `ckptinfo` is a TRANSIENT, caller-owned decode buffer (persist.c's * decode_ckptinfo()), NOT itself an xstring -- j->job.ckptinfo is * xstring-backed (makeString(), xstring.c), so adoption always goes through * makeString() (which copies its input into its own refcounted allocation) * and the existing value is released via dereference() (the RTTI verb), * never free()/FREEUP, mirroring on_job_ckpt()'s own * dereference(j->job.ckptinfo) call. `ckptinfo` itself is always freed here * before returning, regardless of whether adoption happened -- the value * actually stored in j->job.ckptinfo is always a freshly-made makeString() * copy, never the input pointer itself. */ int restore_job_ckpt_state(cluster_t* c, uint64_t jobid, uint32_t state, char* ckptinfo) { job_t* j = NULL; lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&j); unlock(c->jobmgr); if (!j) { FREEUP(ckptinfo); return -ENOENT; } lock(j, RTTI_LOCK_WRITE); j->job.state = (int)state; if (ckptinfo != NULL) { if (j->job.ckptinfo != NULL) dereference(j->job.ckptinfo); j->job.ckptinfo = makeString(ckptinfo); } unlock(j); FREEUP(ckptinfo); return 0; } /* bug-1692: WAL_JOB_PREASON's own sibling of restore_job_ckpt_state() above, * for apply_wal_entry()'s standby catch-up / replicated-apply seam. * * has_field distinguishes a pre-fix bare {jobid,state} payload (has_field=0, * pend_reasons left untouched -- full backward compatibility) from a * post-fix payload that explicitly carries pend_reasons' new value * (has_field=1): preason non-NULL adopts a freshly-decoded value, preason * NULL means the field was genuinely cleared (do_clr_preason) and * pend_reasons should be reset to NULL too -- pend_reasons is the one * field-parity fact type in this family (bug-1678/1687/1689/1690/1691) with * a real CLEAR operation, so unlike restore_job_ckpt_state()/ * restore_job_rusage_state() a NULL decoded value is NOT automatically * "nothing to adopt". * * preason (when non-NULL) is always freshly heap-allocated by * decode_jobpreason() (persist.c) -- plain calloc()/strdup() throughout * (verified via erase_job_preason()'s FREEUP-only release and every * add_*_preason() constructor in src/lib/preason.c, plus the * capnp-generated decode_HostBasedReason/decode_ResourceBasedReason using * STRING_DUP+calloc), NOT xstring/RTTI-backed like ckptinfo or rusage's * host_name -- so a straightforward free-old-then-adopt-new via * free_JobPendReason_ptr() is safe here, no merge-not-replace treatment * needed (unlike bug-1690's WAL_JOB_RUSAGE). Ownership of preason always * transfers to job->job.pend_reasons on success; released here if the job * is gone. */ int restore_job_preason_state(cluster_t* c, uint64_t jobid, uint32_t state, int has_field, xce_job_preason_t* preason) { job_t* j = NULL; lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&j); unlock(c->jobmgr); if (!j) { if (preason != NULL) free_JobPendReason_ptr(&preason); return -ENOENT; } lock(j, RTTI_LOCK_WRITE); j->job.state = (int)state; if (has_field) { if (j->job.pend_reasons != NULL) free_JobPendReason_ptr(&(j->job.pend_reasons)); j->job.pend_reasons = preason; /* NULL (cleared) or the decoded value */ } unlock(j); return 0; } static int get_nslots_in_alloc(xce_host_alloc_t* alloc) { int i; for (i = 0; i < alloc->nallocs; i++) { if (strcmp(alloc->allocs[i]->resname, "slots") == 0) { return (int)alloc->allocs[i]->value; } } return 1; } static void free_launchinfo(xce_launch_t* launch) { int i; for (i = 0; i < launch->nhosts; i++) { FREEUP(launch->hosts[i]); } FREEUP(launch->hosts); } static int mk_launchinfo(xce_launch_t* launch, xce_job_alloc_t* alloc) { int i; int rc = 0; memset(launch, 0x0, sizeof(*launch)); launch->hosts = (xce_host_launch_t**)calloc(alloc->nhosts, sizeof(xce_host_launch_t*)); if (launch->hosts == NULL) { return -ENOMEM; } launch->nhosts = alloc->nhosts; for (i = 0; i < alloc->nhosts; i++) { launch->hosts[i] = (xce_host_launch_t*)calloc(1, sizeof(xce_host_launch_t)); if (launch->hosts[i] == NULL) { rc = -ENOMEM; break; } launch->hosts[i]->host = alloc->allocs[i]->host; launch->hosts[i]->nranks = get_nslots_in_alloc(alloc->allocs[i]); } if (rc < 0) { free_launchinfo(launch); } return rc; } /* * scheduler-dispatch-commit task 1.1 (Req 4.1, 4.2): classifies one of * dispatch_a_job_()'s six post-lookup failure points as retryable * (transient/environmental -- return the job to PEND for a later retry) or * terminal (a locally invalid decision that retrying will not fix -- resolve * the job to a terminal failed state). Declared static/file-local: nothing * outside job.c consumes this classification (design.md's own placement- * correction note). dispatch_a_job_() itself is wired to call this at each * of its six failure points by a later task; this task only adds the * disconnected, independently-testable classification step -- * dispatch_a_job_() below is unmodified by this task. */ typedef enum { XCE_DISPATCH_FAIL_SITE_WORKER_LOOKUP = 0, /* workermgr key_find() rc<0 */ XCE_DISPATCH_FAIL_SITE_WORKER_NULL = 1, /* worker not found (w==NULL) */ XCE_DISPATCH_FAIL_SITE_WORKER_NOT_READY = 2, /* E_XCE_BAD_WORKER */ XCE_DISPATCH_FAIL_SITE_INVALID_ALLOC = 3, /* nc <= 0 */ XCE_DISPATCH_FAIL_SITE_QUEUE_EXEC = 4, /* get_queue_exec() failure */ XCE_DISPATCH_FAIL_SITE_RPC_FAILURE = 5, /* xce_startjob_req() rc<0 */ } xce_dispatch_fail_site_t; typedef enum { XCE_DISPATCH_FAIL_RETRYABLE = 0, XCE_DISPATCH_FAIL_TERMINAL = 1, } xce_dispatch_fail_class_t; /* Keyed on WHICH failure point was hit (site identity), not on the raw rc * that point returned: workermgr's key_find() and get_queue_exec() are two * independently-implemented failure sources not guaranteed to return * disjoint error codes (e.g. both could plausibly return -ENOENT for * unrelated reasons), so classifying by rc value alone would be ambiguous * whenever two sites happen to collide. * * Every named site is handled explicitly by its own case. The `default:` * arm below exists for the impossible-by-construction case of a `site` with * no matching enumerator (every current/planned call site passes a literal * enumerator, so this can only be reached by a genuine programming-invariant * violation, not by any operational/environmental condition) -- a plain * default-less switch does NOT get compiler-enforced exhaustiveness in C * (verified directly: GCC emits only a non-fatal `-Wreturn-type` "control * reaches end of non-void function" warning here, and this project sets no * -Werror anywhere), so falling off the end would silently return an * indeterminate value instead. This mirrors persist.c's wal_classify() * default arm exactly: ERROR-log and return a safe-side classification * (TERMINAL -- never risk an unbounded retry loop on an unrecognized site) * unconditionally, and additionally assert() only under XCE_TESTING, so * xce_controller itself (built without -DXCE_TESTING) can never self- * terminate on this path (CLAUDE.md's controller fail-loud/never-self- * terminate discipline: assert()/abort is reserved for exactly this kind of * impossible-state bug, but production must still never exit on it), while * controller_unit_test (built WITH -DXCE_TESTING) crashes loud and * immediately if it's ever hit. */ static xce_dispatch_fail_class_t xce_classify_dispatch_fail(xce_dispatch_fail_site_t site) { switch (site) { case XCE_DISPATCH_FAIL_SITE_WORKER_LOOKUP: case XCE_DISPATCH_FAIL_SITE_WORKER_NULL: case XCE_DISPATCH_FAIL_SITE_WORKER_NOT_READY: case XCE_DISPATCH_FAIL_SITE_RPC_FAILURE: return XCE_DISPATCH_FAIL_RETRYABLE; case XCE_DISPATCH_FAIL_SITE_INVALID_ALLOC: case XCE_DISPATCH_FAIL_SITE_QUEUE_EXEC: return XCE_DISPATCH_FAIL_TERMINAL; default: ERROR("xce_classify_dispatch_fail: unrecognized failure site %d -- " "TERMINAL fallback; add it to the classification table", (int)site); #ifdef XCE_TESTING assert(0 && "unrecognized dispatch-fail site -- add it to " "xce_classify_dispatch_fail"); #endif return XCE_DISPATCH_FAIL_TERMINAL; } } #ifdef XCE_TESTING /* Test-only accessor: xce_classify_dispatch_fail() and its enum types are * intentionally undeclared in any header (nothing outside job.c consumes * them), so the unit test reaches it through this thin, int-typed wrapper * instead -- mirroring job_array_apply_test_set_force_fail_at()'s existing * XCE_TESTING-only-export pattern earlier in this file (xce_controller * itself is built without -DXCE_TESTING, so this never ships). */ int xce_classify_dispatch_fail_test(int site) { return (int)xce_classify_dispatch_fail((xce_dispatch_fail_site_t)site); } #endif /* XCE_TESTING */ static int dispatch_a_job_(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) { int rc = 0; job_t* j = NULL; worker_t* w = NULL; xce_host_alloc_t** allocs; int nc; xce_job_exec_t einfo; xce_launch_t launchinfo; char jobidStr[64]; /* bug-1588: fmt_jobid() renders the raw idgen()-packed jobid in the * same human-readable "cluster-apiserver-timestamp-sequence" form * already used everywhere else this codebase logs/displays a jobid * (scheduler.c/handler.c/worker's process_manager.c, xcecli's job * listing, waldump) -- job.c's own TRACE lines were the one place * still printing the raw 64-bit packed integer. */ fmt_jobid(jobidStr, sizeof(jobidStr), alloc->jobid); TRACE("%s jobid %s", __func__, jobidStr); lock(c->jobmgr, RTTI_LOCK_READ); rc = key_find(c->jobmgr, (void*)alloc->jobid, (void**)&j); if (j != NULL) { reference(j); } unlock(c->jobmgr); if (rc < 0) { return rc; } if (j == NULL) { PERROR("jobmgr_lockjob"); return -errno; } nc = alloc->nhosts; if (nc <= 0) { /* scheduler-dispatch-commit task 3.2 (Req 4.2, 4.5): resolves this * terminal (locally-invalid) dispatch failure -- j's own reference * held since the lookup above is released via the dereference(j) * call below exactly as it already was before this task (not * duplicated). Nothing was ever committed to j->job.allocs on this * path (task 3.1 gates that commit on a successful * xce_startjob_req(), never reached here), so Req 4.5's "release * any tentative allocation data" is satisfied vacuously by * construction -- there is nothing to release. */ xce_dispatch_fail_class_t dfclass; job_failinfo_t failinfo = { .jobid = alloc->jobid }; lock(j, RTTI_LOCK_WRITE); dfclass = xce_classify_dispatch_fail(XCE_DISPATCH_FAIL_SITE_INVALID_ALLOC); switch (dfclass) { case XCE_DISPATCH_FAIL_TERMINAL: break; case XCE_DISPATCH_FAIL_RETRYABLE: default: /* Impossible per today's classification table (confirmed: * xce_classify_dispatch_fail() returns TERMINAL for this site * unconditionally) -- task 3.3 is what wires real retryable * handling at the other four sites, not this one. Fail loud * and fall through to the same terminal resolution as a safe * fallback, mirroring xce_classify_dispatch_fail()'s own * default: convention. */ ERROR("dispatch_a_job_: XCE_DISPATCH_FAIL_SITE_INVALID_ALLOC " "classified non-terminal (%d) -- impossible per the " "current classification table; resolving terminal as a " "safe fallback", (int)dfclass); #ifdef XCE_TESTING assert(0 && "XCE_DISPATCH_FAIL_SITE_INVALID_ALLOC classified " "non-terminal -- update the classification table or " "this call site"); #endif break; } failinfo.code = -EINVAL; failinfo.detail = "dispatch failed: invalid allocation"; mark_job_fail(c, j, &failinfo); unlock(j); dereference(j); return -EINVAL; } /* FIXME: change case of nc > 1 just hard coded mpi as * the real job type later */ mk_launchinfo(&launchinfo, alloc); allocs = (xce_host_alloc_t**)alloc->allocs; /* using head node of job */ lock(c->workermgr, RTTI_LOCK_READ); rc = key_find(c->workermgr, (void*)allocs[0]->host, (void**)&w); if (w != NULL) { reference(w); } unlock(c->workermgr); if (rc < 0) { /* scheduler-dispatch-commit task 3.3 (Req 4.1, 4.3, 4.4, 4.5): * resolves this retryable (transient/environmental) dispatch * failure -- closes bug-1627's leak site #1 (no dereference(j) * existed on this path before this task). rc already holds the * stable natural error code from key_find() above; no further call * touches it before it is returned below. Nothing was ever * committed to j->job.allocs on this path (task 3.1 gates that * commit on a successful xce_startjob_req(), never reached here), * so Req 4.4/4.5's "release any tentative allocation data" is * satisfied vacuously by construction. */ xce_dispatch_fail_class_t dfclass; job_failinfo_t failinfo = { .jobid = alloc->jobid }; bool resolved_retryable = false; uint32_t failcount; lock(j, RTTI_LOCK_WRITE); dfclass = xce_classify_dispatch_fail( XCE_DISPATCH_FAIL_SITE_WORKER_LOOKUP); switch (dfclass) { case XCE_DISPATCH_FAIL_RETRYABLE: failcount = ++j->dispatch_fail_count; if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) { resolved_retryable = true; } else { failinfo.code = rc; failinfo.detail = "dispatch failed: worker lookup failed " "repeatedly, retry bound exceeded"; mark_job_fail(c, j, &failinfo); } break; case XCE_DISPATCH_FAIL_TERMINAL: default: /* Impossible per today's classification table (confirmed: * xce_classify_dispatch_fail() returns RETRYABLE for this site * unconditionally) -- fail loud and fall through to the same * terminal resolution as a safe fallback, mirroring * xce_classify_dispatch_fail()'s own default: convention. */ ERROR("dispatch_a_job_: XCE_DISPATCH_FAIL_SITE_WORKER_LOOKUP " "classified TERMINAL (%d) -- impossible per the current " "classification table; resolving terminal as a safe " "fallback", (int)dfclass); #ifdef XCE_TESTING assert(0 && "XCE_DISPATCH_FAIL_SITE_WORKER_LOOKUP classified " "terminal -- update the classification table or " "this call site"); #endif failinfo.code = rc; failinfo.detail = "dispatch failed: worker lookup failed"; mark_job_fail(c, j, &failinfo); break; } unlock(j); dereference(j); if (resolved_retryable) { notify_dispatch(c, ep, alloc->jobid, rc); } return rc; } if (w == NULL) { int dfrc; xce_dispatch_fail_class_t dfclass; job_failinfo_t failinfo = { .jobid = alloc->jobid }; bool resolved_retryable = false; uint32_t failcount; /* scheduler-dispatch-commit task 3.3: capture errno FIRST, before * any other call (free_launchinfo() included) -- errno is not * guaranteed stable across subsequent calls. */ dfrc = -errno; free_launchinfo(&launchinfo); /* Req 4.1, 4.3, 4.4, 4.5: resolves this retryable (transient/ * environmental) dispatch failure -- closes bug-1627's leak site * #2 (no dereference(j) existed on this path before this task). */ lock(j, RTTI_LOCK_WRITE); dfclass = xce_classify_dispatch_fail( XCE_DISPATCH_FAIL_SITE_WORKER_NULL); switch (dfclass) { case XCE_DISPATCH_FAIL_RETRYABLE: failcount = ++j->dispatch_fail_count; if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) { resolved_retryable = true; } else { failinfo.code = dfrc; failinfo.detail = "dispatch failed: worker not found " "repeatedly, retry bound exceeded"; mark_job_fail(c, j, &failinfo); } break; case XCE_DISPATCH_FAIL_TERMINAL: default: ERROR("dispatch_a_job_: XCE_DISPATCH_FAIL_SITE_WORKER_NULL " "classified TERMINAL (%d) -- impossible per the current " "classification table; resolving terminal as a safe " "fallback", (int)dfclass); #ifdef XCE_TESTING assert(0 && "XCE_DISPATCH_FAIL_SITE_WORKER_NULL classified " "terminal -- update the classification table or " "this call site"); #endif failinfo.code = dfrc; failinfo.detail = "dispatch failed: worker not found"; mark_job_fail(c, j, &failinfo); break; } unlock(j); dereference(j); if (resolved_retryable) { notify_dispatch(c, ep, alloc->jobid, dfrc); } return dfrc; } lock(w, RTTI_LOCK_READ); if ((w->ep == NULL) || (w->host.status != 0)) { int dfrc; xce_dispatch_fail_class_t dfclass; job_failinfo_t failinfo = { .jobid = alloc->jobid }; bool resolved_retryable = false; uint32_t failcount; ERROR("worker %s is not in correct state, status %x ep %p", w->host.name, w->host.status, w->ep); free_launchinfo(&launchinfo); SET_ERROR(E_XCE_BAD_WORKER); /* scheduler-dispatch-commit task 3.3: capture errno immediately * after SET_ERROR(), before any further call -- errno is not * guaranteed stable across unlock(w)/dereference(w)/lock(j)/etc. */ dfrc = -errno; unlock(w); dereference(w); /* Req 4.1, 4.3, 4.4, 4.5: resolves this retryable (transient/ * environmental) dispatch failure -- closes bug-1627's leak site * #3 (no dereference(j) existed on this path before this task). */ lock(j, RTTI_LOCK_WRITE); dfclass = xce_classify_dispatch_fail( XCE_DISPATCH_FAIL_SITE_WORKER_NOT_READY); switch (dfclass) { case XCE_DISPATCH_FAIL_RETRYABLE: failcount = ++j->dispatch_fail_count; if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) { resolved_retryable = true; } else { failinfo.code = dfrc; failinfo.detail = "dispatch failed: worker not ready " "repeatedly, retry bound exceeded"; mark_job_fail(c, j, &failinfo); } break; case XCE_DISPATCH_FAIL_TERMINAL: default: ERROR("dispatch_a_job_: " "XCE_DISPATCH_FAIL_SITE_WORKER_NOT_READY classified " "TERMINAL (%d) -- impossible per the current " "classification table; resolving terminal as a safe " "fallback", (int)dfclass); #ifdef XCE_TESTING assert(0 && "XCE_DISPATCH_FAIL_SITE_WORKER_NOT_READY classified " "terminal -- update the classification table or " "this call site"); #endif failinfo.code = dfrc; failinfo.detail = "dispatch failed: worker not ready"; mark_job_fail(c, j, &failinfo); break; } unlock(j); dereference(j); if (resolved_retryable) { notify_dispatch(c, ep, alloc->jobid, dfrc); } return dfrc; } /* FIXME: we may consider to optimize when a job has enter a queue? */ if (1) { char* q_pre_exec; char* q_post_exec; rc = get_queue_exec(c, j->job.submit->queue, &q_pre_exec, &q_post_exec); if (rc < 0) { /* scheduler-dispatch-commit task 3.2 (Req 4.2, 4.5): resolves * this terminal (locally-invalid) dispatch failure -- j's own * reference held since the lookup above is released via the * dereference(j) call below exactly as it already was before * this task (not duplicated). Nothing was ever committed to * j->job.allocs on this path (task 3.1 gates that commit on a * successful xce_startjob_req(), never reached here), so Req * 4.5's "release any tentative allocation data" is satisfied * vacuously by construction. unlock(w)/dereference(w)/ * free_launchinfo(&launchinfo) stay in their existing * positions -- only this new lock/classify/mark_job_fail/ * unlock sequence is inserted around the existing * dereference(j) call. */ xce_dispatch_fail_class_t dfclass; job_failinfo_t failinfo = { .jobid = alloc->jobid }; unlock(w); dereference(w); lock(j, RTTI_LOCK_WRITE); dfclass = xce_classify_dispatch_fail( XCE_DISPATCH_FAIL_SITE_QUEUE_EXEC); switch (dfclass) { case XCE_DISPATCH_FAIL_TERMINAL: break; case XCE_DISPATCH_FAIL_RETRYABLE: default: /* Impossible per today's classification table (confirmed: * xce_classify_dispatch_fail() returns TERMINAL for this * site unconditionally) -- task 3.3 is what wires real * retryable handling at the other four sites, not this * one. Fail loud and fall through to the same terminal * resolution as a safe fallback, mirroring * xce_classify_dispatch_fail()'s own default: convention. */ ERROR("dispatch_a_job_: XCE_DISPATCH_FAIL_SITE_QUEUE_EXEC " "classified non-terminal (%d) -- impossible per the " "current classification table; resolving terminal " "as a safe fallback", (int)dfclass); #ifdef XCE_TESTING assert(0 && "XCE_DISPATCH_FAIL_SITE_QUEUE_EXEC classified " "non-terminal -- update the classification table " "or this call site"); #endif break; } failinfo.code = rc; failinfo.detail = "dispatch failed: queue exec info lookup " "failed"; mark_job_fail(c, j, &failinfo); unlock(j); dereference(j); free_launchinfo(&launchinfo); return rc; } if (j->job.submit->start_info->q_pre_exec != NULL) { } if (j->job.submit->start_info->q_post_exec != NULL) { } j->job.submit->start_info->q_pre_exec = q_pre_exec; j->job.submit->start_info->q_post_exec = q_post_exec; } einfo.jobid = alloc->jobid; einfo.start_info = j->job.submit->start_info; einfo.launchinfo = &launchinfo; rc = xce_startjob_req(w->ep, &einfo); unlock(w); dereference(w); free_launchinfo(&launchinfo); /* scheduler-dispatch-commit task 3.1 (Req 2.1-2.4, 5.1): the allocation * commit -- previously unconditional regardless of xce_startjob_req()'s * outcome -- is now gated on the dispatch to the worker actually * succeeding, and the job's transition to a running state happens in * that same successful outcome. Protected under this component's own * lock(j, RTTI_LOCK_WRITE): dispatch_a_job_() previously took no lock on * `j` itself here (only on c->jobmgr/c->workermgr/w), a real * pre-existing gap this restructuring corrects for this call site * (design.md's Concurrency Strategy note -- not fixed anywhere else in * this file). On failure, state is left untouched (still PEND) and * nothing is committed, so there is nothing to release (Req 4.4/4.5 * satisfied vacuously by construction for this path). The worker's own * later confirmation that the job actually started * (record_job_start_fact()/WAL_JOB_START) continues to fire exactly as * it does today, unaffected by this change. */ if (rc >= 0) { lock(j, RTTI_LOCK_WRITE); if (j->job.allocs != NULL) { free_JobAlloc_ptr(&(j->job.allocs)); } j->job.allocs = (xce_job_alloc_t*)calloc(1, sizeof(xce_job_alloc_t)); j->job.allocs->jobid = alloc->jobid; j->job.allocs->nhosts = nc; MOVE_PTR(j->job.allocs->allocs, allocs); alloc->allocs = NULL; j->job.state = JOB_STAT_RUN; /* bug-1697: start_time is now stamped HERE, synchronously alongside * the RUN transition, rather than left to the worker's later * WAL_JOB_START report. dispatch_a_job_() setting job.state = * JOB_STAT_RUN synchronously means any WAL_JOB_START fact for this * job always finds state already == RUN by the time it applies * (apply_job_update_transition() reports JOB_TRANSITION_NOOP_ * DUPLICATE), so it can never be the genuine first RUN transition * and start_time adoption gated on that transition would be * permanently dead code -- see wal_types.h's WAL_JOB_START doc * comment for the full rationale. */ j->job.start_time = time(NULL); /* Reset the consecutive-retryable-failure counter (task 1.3): a * fresh successful dispatch clears any prior failure history, so a * job that later fails again is evaluated against the bound as if * it had only failed once (Req 4.3; the bound itself is exercised * end-to-end by task 5.2, not here). */ j->dispatch_fail_count = 0; /* Durable record of the dispatch-time RUN transition (task 1.2), * now including the resource allocation itself (bug-1678): the * plain {jobid, state} shape persist_submit_job_fact() writes was * not enough to reconstruct job.allocs on a restart replay, leaving * EXEC_HOST permanently blank for a job that was RUNNING before * shutdown. j->job.allocs is read here, still under this same * write lock, as the freshly-constructed canonical struct just * stored above -- not the (now partially moved-from) local * `alloc`/`allocs` variables. bug-1697: also carries j->job. * start_time, the same snapshot just stamped above under this same * write lock, so a restart replay recovers the true dispatch-time * clock instead of leaving it at 0 forever. */ (void)persist_submit_job_dispatch_fact(&c->persist, alloc->jobid, JOB_STAT_RUN, (uint64_t)j->job.start_time, j->job.allocs); unlock(j); /* bug-1679: the mutation above is live and synchronous -- unlike * record_job_start_fact()'s durable-append-only sibling, this * function never defers to apply_committed_entry() for its own * notification, and by the time that WAL_JOB_DISPATCH entry does * apply, state already matches (a correct, harmless no-op) so its * own hook-fire is silently skipped. Without this explicit call, * the apiserver is never told this job reached RUN. persist.h's * doc comment on persist_notify_job_updated_already_persisted has * the full rationale and the double-WAL-append hazard it guards * against. */ persist_notify_job_updated_already_persisted(c, j); } else { xce_dispatch_fail_class_t dfclass; job_failinfo_t failinfo = { .jobid = alloc->jobid }; bool resolved_retryable = false; uint32_t failcount; SET_ERROR(-rc); PERROR("xce_startjob_req"); /* scheduler-dispatch-commit task 3.3 (Req 4.1, 4.3, 4.4, 4.5): * resolves this retryable (transient/environmental) dispatch * failure -- closes bug-1627's leak site #4 (no dereference(j) * existed on this path before this task). rc already holds the * stable, captured return of xce_startjob_req() -- no call between * that assignment and here has touched it. */ lock(j, RTTI_LOCK_WRITE); dfclass = xce_classify_dispatch_fail( XCE_DISPATCH_FAIL_SITE_RPC_FAILURE); switch (dfclass) { case XCE_DISPATCH_FAIL_RETRYABLE: failcount = ++j->dispatch_fail_count; if (failcount <= XCE_DISPATCH_FAIL_RETRY_BOUND) { resolved_retryable = true; } else { failinfo.code = rc; failinfo.detail = "dispatch failed: xce_startjob_req " "failed repeatedly, retry bound " "exceeded"; mark_job_fail(c, j, &failinfo); } break; case XCE_DISPATCH_FAIL_TERMINAL: default: /* Impossible per today's classification table (confirmed: * xce_classify_dispatch_fail() returns RETRYABLE for this site * unconditionally) -- fail loud and fall through to the same * terminal resolution as a safe fallback, mirroring * xce_classify_dispatch_fail()'s own default: convention. */ ERROR("dispatch_a_job_: XCE_DISPATCH_FAIL_SITE_RPC_FAILURE " "classified TERMINAL (%d) -- impossible per the current " "classification table; resolving terminal as a safe " "fallback", (int)dfclass); #ifdef XCE_TESTING assert(0 && "XCE_DISPATCH_FAIL_SITE_RPC_FAILURE classified " "terminal -- update the classification table or " "this call site"); #endif failinfo.code = rc; failinfo.detail = "dispatch failed: xce_startjob_req failed"; mark_job_fail(c, j, &failinfo); break; } unlock(j); dereference(j); if (resolved_retryable) { notify_dispatch(c, ep, alloc->jobid, rc); } } return rc; } #ifdef XCE_TESTING /* Test-only accessor: dispatch_a_job_() is static/file-local (its only * caller, handle_jobdispatch(), is unmodified by this task) -- mirroring * xce_classify_dispatch_fail_test()'s existing XCE_TESTING-only-export * pattern earlier in this file (xce_controller itself is built without * -DXCE_TESTING, so this never ships). */ int dispatch_a_job_test(cluster_t* c, endpoint_t* ep, xce_job_alloc_t* alloc) { return dispatch_a_job_(c, ep, alloc); } #endif /* XCE_TESTING */ int job_ctrl(cluster_t* c, xce_jobctrl_t* ctrinfo) { uint64_t jobid = ctrinfo->jobid; job_t* j = NULL; int rc = 0; TRACE("%s job_ctrl job %ld", __func__, jobid); lock(c->jobmgr, RTTI_LOCK_READ); rc = key_find(c->jobmgr, (void*)jobid, (void**)&j); if (j != NULL) { reference(j); } unlock(c->jobmgr); if (rc < 0) { return rc; } if (j == NULL) { PERROR("jobmgr_lockjob"); return -ENOENT; } /* bug-447: key_find returns 1 on found (binary librtti, unnormalizable * upstream). Normalize the found-rc to 0 so the not-started siginfo * branch below (which appends via the mark_job_* hooks but never * reassigns rc) returns 0 to the handler's commit-barrier pattern * instead of leaking 1 as the reply status. Real errors are * untouched: rc < 0 returned above, the j == NULL miss returned * -ENOENT, and every other branch sets rc itself. */ rc = 0; /* FIXME: eliminate held multiple locks */ lock(j, RTTI_LOCK_READ); if (j->job.allocs != NULL) { xce_host_alloc_t** allocs = j->job.allocs->allocs; worker_t* w = NULL; /* job control command will be sent to first node * for parallel jobs */ lock(c->workermgr, RTTI_LOCK_READ); rc = key_find(c->workermgr, (void*)allocs[0]->host, (void**)&w); if (w != NULL) { reference(w); } unlock(c->workermgr); if (rc < 0) { unlock(j); dereference(j); return rc; } if (w != NULL) { lock(w, RTTI_LOCK_READ); /* FIXME: better method to check host status */ if ((w->ep != NULL) && (w->host.status == 0)) { rc = xce_jobctrl_req(w->ep, ctrinfo); if (rc < 0) { ERROR("job ctrl %ld on host %s returns %d", jobid, w->host.name, rc); } } else { ERROR("worker %s is not in correct state, status %x ep %p", w->host.name, w->host.status, w->ep); rc = -errno; } unlock(w); dereference(w); } else { rc = -errno; } } else { /* job has not been started yet */ job_failinfo_t failinfo = { .jobid = jobid, }; xce_use(&failinfo); switch (ctrinfo->action) { case JobControl_u_siginfo: switch (ctrinfo->u.siginfo->signo) { case SIGCONT: if (j->job.state & (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP)) { int state = j->job.state; /* resume suspend job? */ state &= ~(JOB_STAT_PSUSP | JOB_STAT_USUSP); mark_job_update(c, j, state); } break; case SIGSTOP: case SIGTSTP: if (!(j->job.state & (JOB_STAT_PSUSP | JOB_STAT_USUSP | JOB_STAT_SSUSP))) { int state = j->job.state; state |= JOB_STAT_PSUSP; mark_job_update(c, j, state); } break; case SIGKILL: case SIGTERM: default: if (j->job.state & (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) { break; } /* FIXME: */ failinfo.code = 128 + ctrinfo->u.siginfo->signo; failinfo.detail = "Job killed by signal"; mark_job_fail(c, j, &failinfo); break; } break; case JobControl_u_ckptoption: if (j->job.state & (JOB_STAT_DONE | JOB_STAT_EXIT | JOB_STAT_ERR)) { ERROR("job <%ld> has been finished, no checkpoint can be done.", jobid); } else { ERROR("job %ld has not been started, no checkpoint can be done", jobid); } rc = -EINVAL; break; default: rc = -EINVAL; break; } } unlock(j); return rc; } static void notify_dispatch(cluster_t* c, endpoint_t* ep, uint64_t jobid, int code) { /* clang-format off */ job_failinfo_t info = { .jobid = jobid, .code = code }; xce_notify_data_t data = { .event = NotifyData_data_jobDispatchFail, .data = { .dispatchfail = &info } }; /* clang-format on */ notify(c, ep, &data); } int handle_jobdispatch(cluster_t* c, endpoint_t* ep, xce_request_t* request) { int rc = 0; int i; xce_job_dispatch_t* dispatchmsg = request->body.dispatchmsg; xce_reply_t reply; /* reply job dispatch that we'll handle request */ init_reply_by_req(&reply, request); reply.status = 0; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); for (i = 0; i < dispatchmsg->njobs; i++) { xce_job_alloc_t* alloc = NULL; MOVE_PTR(alloc, dispatchmsg->allocs[i]); /* scheduler-dispatch-commit task 4.1 (Req 3.1, 4.1, 4.2): reporting * the dispatch outcome is now entirely owned by dispatch_a_job_() * itself (tasks 3.1-3.3) -- it already calls notify_dispatch() on * the retryable-failure path and relies on mark_job_fail()'s own * hook chain for the terminal-failure path, never on success. This * caller used to unconditionally call notify_dispatch() again for * every job regardless of that outcome, producing a conflicting * double report (e.g. both jobDispatchFail and schedFailJob for the * same terminal failure). The return value has nothing left for * this caller to act on. */ (void)dispatch_a_job_(c, ep, alloc); free_JobAlloc_ptr(&(alloc)); } return rc; } /* raft-apply-decoupling task 3.2: shared state for one handle_jobsubmit() * call's asynchronously-resolving per-entry commit continuations. Exactly * one instance per client request — an array submission registers one * continuation per array element, all sharing this struct, so the client * still gets exactly one reply for the whole batch (matching pre-3.2 * per-request reply semantics), never one reply per array element. * * Lifetime: allocated in handle_jobsubmit() before it returns; released by * whichever per-entry continuation (jobsubmit_commit_continuation) or * apply-complete callback (jobsubmit_apply_complete) observes `pending` * reach 0 — the last one out closes the door. * * IMPORTANT: a COMMITTED outcome does NOT decrement `pending` by itself * (see jobsubmit_commit_continuation below) — task 3.2's own contract is * that a committed entry is handed off to the not-yet-built ordered apply * pump (task 6.1/6.2), and only jobsubmit_apply_complete (invoked BY that * future pump) decrements `pending` for it. A batch with any committed * entry is therefore only ever freed once task 6 exists and actually * applies it. This mirrors the design's already-accepted "durably * committed, not yet applied" posture for retained facts during quorum * loss (Req 6.1) — it is not a leak. */ typedef struct { cluster_t* c; endpoint_t* ep; /* ref'd via endpoint_ref(); released exactly * once, in jobsubmit_async_ctx_release() */ uint64_t seq; /* xce_request_t.seq, copied for the reply */ uint32_t opcode; /* xce_request_t.opcode, copied for the reply */ uint64_t jobid; /* base jobid of the (possibly array) submission, * echoed back on the success reply (task 6.2) -- * matches pre-3.1 semantics (git history commit * 6f1fa97), which always reported the BASE id, * never a per-element id */ uint64_t array_size; /* bug-1419: paired with jobid to remove the * matching inflight_range_add() entry on * release -- see jobid_range_exists' doc * comment for the full contract */ pthread_mutex_t lock; /* guards pending/replied below */ int pending; /* entries not yet FULLY resolved (see above) */ int replied; /* 1 once a reply has been sent for this batch */ } jobsubmit_async_ctx_t; static void jobsubmit_async_ctx_release(jobsubmit_async_ctx_t* actx) { /* bug-1419: this is the single choke point every resolution path * (commit failure, apply completion, and the OOM/alloc-failure * early-outs in handle_jobsubmit below) already reaches -- clearing * the in-flight registration here means a retry is only ever * dedup-blocked while this exact submission's outcome is genuinely * still unresolved, never after. */ inflight_range_remove(actx->jobid, actx->array_size); endpoint_unref(actx->ep); pthread_mutex_destroy(&actx->lock); free(actx); } /* Sends exactly one failure reply for the whole batch — the first failure * observed wins; later failures/hand-offs never send a second, possibly * contradictory reply for the same original request (Req 9.3, read at the * per-REQUEST granularity a job-array submission actually has one of). */ static void jobsubmit_reply_failure_once(jobsubmit_async_ctx_t* actx, int rc) { xce_reply_t reply; int send = 0; pthread_mutex_lock(&actx->lock); if (!actx->replied) { actx->replied = 1; send = 1; } pthread_mutex_unlock(&actx->lock); if (!send) return; memset(&reply, 0, sizeof(reply)); reply.seq = actx->seq; reply.opcode = actx->opcode; reply.status = rc; reply.data_which = RequestReply_data_nodata; xce_reply(actx->ep, actx->c->myhost, &reply); } /* Sends exactly one success reply for the whole batch, carrying the base * jobid — the sibling of jobsubmit_reply_failure_once, gated on the SAME * actx->replied flag (Req 9.3) so a batch that already sent a failure reply * for a sibling element never follows it with a contradictory success reply, * and vice versa. Only ever reached from jobsubmit_apply_complete once every * element's continuation has resolved (raft-apply-decoupling task 6.2). */ static void jobsubmit_reply_success_once(jobsubmit_async_ctx_t* actx) { xce_reply_t reply; int send = 0; pthread_mutex_lock(&actx->lock); if (!actx->replied) { actx->replied = 1; send = 1; } pthread_mutex_unlock(&actx->lock); if (!send) return; memset(&reply, 0, sizeof(reply)); reply.seq = actx->seq; reply.opcode = actx->opcode; reply.status = 0; reply.data_which = RequestReply_data_jobid; reply.data.jobid = actx->jobid; xce_reply(actx->ep, actx->c->myhost, &reply); } /* raft-apply-decoupling task 6 hand-off target: invoked by the ordered apply * pump (persist_apply_pump_run(), task 6.1/6.2) immediately after it applies * this entry via apply_committed_entry() — task 6.2's own contract, "fire a * command's registered continuation immediately after its entry applies". * See jobsubmit_commit_continuation's COMMITTED branch below, which * registers this as the hand-off target instead of applying or replying * itself. * * apply_committed_entry()'s WAL_JOB_SUBMIT case (persist.c) already * constructs the job (via restore_job()) AND fires the HOOK_DETAIL_JOB_NEW * AFTER_EVENT notification, independently per entry, strictly BEFORE the * apply pump reaches the persist_take_pending_apply()/invoke step that calls * this function — so by the time this function runs, the job is already * visible in c->jobmgr and already notified. This function therefore only * sends the batch's success reply (echoing the base jobid) once every entry * has been applied; it never reconstructs or re-notifies anything — see the * removed-code note above handle_jobsubmit's old (pre-3.2) construct/notify * block for the full reasoning (git history, commit 6f1fa97, task 3.1). */ static void jobsubmit_apply_complete(void* ctx, uint64_t lsn) { jobsubmit_async_ctx_t* actx = (jobsubmit_async_ctx_t*)ctx; int done; (void)lsn; pthread_mutex_lock(&actx->lock); actx->pending--; done = (actx->pending <= 0); pthread_mutex_unlock(&actx->lock); if (done) { jobsubmit_reply_success_once(actx); jobsubmit_async_ctx_release(actx); } } /* raft-apply-decoupling task 3.2: per-entry outcome of one array element's * commit_submit_async() registration (persist_submit_job_payloads_async's * `fn` callback). * * On FAILED/timed-out (rc != 0): reply failure now — no apply is needed on * this path (Req 2.3). This entry has nothing further to wait on, so it * decrements `pending` itself. * * On COMMITTED (rc == 0): task 3.2 must NOT apply the job or reply here * (the task's own explicit instruction) — hand the entry off to the * not-yet-built ordered apply pump (task 6.1/6.2) via * persist_register_pending_apply(), the producer-side hand-off this task * builds. `pending` is intentionally NOT decremented here for a committed * entry — see jobsubmit_async_ctx_t's doc comment. */ static void jobsubmit_commit_continuation(void* ctx, uint64_t lsn, int rc) { jobsubmit_async_ctx_t* actx = (jobsubmit_async_ctx_t*)ctx; if (rc != 0) { int done; WARN("handle_jobsubmit: lsn=%lu commit failed/timed out (rc=%d) — " "rejecting; no job ever constructed for it", (unsigned long)lsn, rc); jobsubmit_reply_failure_once(actx, rc); pthread_mutex_lock(&actx->lock); actx->pending--; done = (actx->pending <= 0); pthread_mutex_unlock(&actx->lock); if (done) jobsubmit_async_ctx_release(actx); return; } /* COMMITTED: hand off to task 6's ordered apply pump instead of * applying/replying directly from this continuation — see * persist_register_pending_apply's doc comment (persist.h) for the * full hand-off contract. */ persist_register_pending_apply(&actx->c->persist, lsn, jobsubmit_apply_complete, actx); } /* job-array-single-wal-record task 2.2 (design.md §WAL Array Submission): * the array-shaped sibling of handle_jobsubmit()'s single-job append + * continuation-registration block below, taken for submit->array_size > 1. * Called strictly AFTER handle_jobsubmit() has already run the SAME * pre-commit range validation (jobid_range_exists + inflight_range_try_add) * the single-job path uses — this function never repeats or reorders that * validation (Req 9.1, 9.2); it only takes over at the point the two paths * diverge. * * jobsubmit_async_ctx_t / jobsubmit_commit_continuation / jobsubmit_apply_ * complete / jobsubmit_reply_success_once / jobsubmit_reply_failure_once are * reused completely unmodified — task 2.1's persist_submit_job_array_async() * plugs into the exact same commit-continuation contract as persist_submit_ * job_payloads_async() does for the single-job path (persist.h). The only * difference from the single-job construction below is actx->pending = 1 * instead of one entry per array element (Req 1.1-1.3): exactly one * WAL_JOB_ARRAY_SUBMIT record describes the whole array, so exactly one * outcome must resolve before the batch's single reply fires (Req 9.3). No * job_t/xce_job_t payload is built here — array expansion into individual * job entries happens later, in the shared apply seam (task 3.x), not on * this call path. */ static int handle_jobsubmit_array(cluster_t* c, endpoint_t* ep, xce_request_t* request, uint64_t jobid, xce_job_submit_t* submit) { xce_reply_t reply; uint64_t array_lsn = 0; jobsubmit_async_ctx_t* actx; init_reply_by_req(&reply, request); actx = calloc(1, sizeof(*actx)); if (actx == NULL) { inflight_range_remove(jobid, submit->array_size); reply.status = -ENOMEM; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); return -ENOMEM; } actx->c = c; actx->ep = ep; actx->seq = request->seq; actx->opcode = request->opcode; actx->jobid = jobid; actx->array_size = submit->array_size; actx->pending = 1; /* one WAL_JOB_ARRAY_SUBMIT record, one outcome */ actx->replied = 0; pthread_mutex_init(&actx->lock, NULL); endpoint_ref(ep); /* Mirrors handle_jobsubmit's own call below: the return value carries * no information this function still needs to act on — fn (jobsubmit_ * commit_continuation) has already been invoked, or is already * guaranteed to be invoked later, exactly once, by the time this call * returns (persist_submit_job_array_async's own documented contract, * persist.h). */ (void)persist_submit_job_array_async(&c->persist, jobid, submit->array_size, submit, &array_lsn, jobsubmit_commit_continuation, actx); return 0; } 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 plist; /* xce_job_t* WAL payloads, pre-commit */ uint64_t* lsns = NULL; int rc; submit = request->body.jobsubmsg; init_reply_by_req(&reply, request); if (NULLSTR(submit->submit->user) || NULLSTR(submit->submit->queue) || (submit->submit->start_info == NULL) || NULLSTR(submit->submit->start_info->cmdline)) { errno = EINVAL; reply.status = -errno; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); return -errno; } 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; } /* bug-1419: atomically claim this jobid range as in-flight, closing the * window jobid_range_exists' c->jobmgr check alone cannot see (see its * doc comment). A false return means another submission (a genuine * concurrent collision, or -- the actual motivating case -- this exact * request's own retried resend racing its still-in-flight original * attempt) already holds this range; reject exactly as the * jobid_range_exists branch above does. Released in * jobsubmit_async_ctx_release() once this submission's outcome is * fully resolved; the two error returns below (which occur BEFORE * actx exists to carry that release) remove it directly. */ if (!inflight_range_try_add(jobid, submit->array_size)) { errno = EEXIST; reply.status = -errno; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); return -errno; } /* job-array-single-wal-record task 2.2 (Req 1.1-1.3, 8.1, 9.1, 9.2): * array_size == 1 falls through to today's single-job path below, * completely unmodified (Req 8.1) — array_size > 1 diverges here, * strictly AFTER the same pre-commit range validation above, into the * single-record append path task 2.1 built. */ if (submit->array_size > 1) return handle_jobsubmit_array(c, ep, request, jobid, submit); /* 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); inflight_range_remove(jobid, submit->array_size); 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; converted to the continuation mechanism by task 3.2): register * every payload's commit outcome instead of blocking on the majority * barrier. This handler returns as soon as every payload is durably * appended locally and its outcome-resolution path is established — * BEFORE any of them necessarily commit. From this point on, job * construction and the client reply happen from * jobsubmit_commit_continuation / jobsubmit_apply_complete (above), * never again from this stack frame. * * REMOVED (raft-apply-decoupling task 3.2) relative to task 3.1's * version of this function: the post-commit create_job()-loop + * jlist-build + xce_run_hooks(AFTER_EVENT, HOOK_DETAIL_JOB_NEW) block * (bug-1408's fix: unconditional per-successfully-constructed-element * notification) is gone, not merely deferred. It is SUPERSEDED, not * resurrected, by apply_committed_entry()'s existing WAL_JOB_SUBMIT * case (persist.c, raft-apply-decoupling task 2.2, already built): * that function already does restore_job() + fires HOOK_DETAIL_JOB_NEW * per entry, independently per array element — which is structurally * immune to the bug-1408 class (a sibling's apply failure can never * suppress another sibling's notification, because each LSN applies * through its own independent call). The full reasoning is preserved * in this task's status report; the removed code itself remains in * git history at commit 6f1fa97. Whoever builds task 6.1/6.2 should * call apply_committed_entry() (not resurrect this loop) and should * expect jobsubmit_apply_complete to need only the success reply. */ jobsubmit_async_ctx_t* actx = calloc(1, sizeof(*actx)); if (actx == NULL) { free_job_submit_payloads(&plist); free(lsns); inflight_range_remove(jobid, submit->array_size); reply.status = -ENOMEM; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); return -ENOMEM; } actx->c = c; actx->ep = ep; actx->seq = request->seq; actx->opcode = request->opcode; actx->jobid = jobid; actx->array_size = submit->array_size; actx->pending = plist.count; actx->replied = 0; pthread_mutex_init(&actx->lock, NULL); endpoint_ref(ep); (void)persist_submit_job_payloads_async(&c->persist, &plist, lsns, jobsubmit_commit_continuation, actx); /* Every payload's WAL data is no longer needed once appended: the * future apply pump (task 6) re-reads each committed entry's payload * from the WAL itself — the same way restart replay already does — * not from this in-memory copy. persist_submit_job_payloads_async has * ALREADY invoked jobsubmit_commit_continuation, synchronously or * asynchronously, exactly once for every payload by (or after) the * time it returns; its own return value carries no information this * function still needs to act on. */ free_job_submit_payloads(&plist); free(lsns); /* The calling thread returns here, before any payload's commit outcome * is necessarily known (raft-apply-decoupling task 3.2, Req 1.1, 1.4, * 2.1, 2.3) — every reply from this point on is sent by * jobsubmit_reply_failure_once or (once task 6 exists) * jobsubmit_apply_complete, off this stack frame. */ return 0; } /* 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; 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) { 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 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 JOB_TRANSITION_REJECTED_BACKWARD; } /* update job information */ job->job.state = JOB_STAT_RUN; job->job.start_time = time(NULL); if (job->job.pidinfo == NULL) { job->job.pidinfo = (job_pidinfo_t*)calloc(1, sizeof(job_pidinfo_t)); } job->job.pidinfo->npids = copy_int_list((int**)&(job->job.pidinfo->pids), (int*)info->pids, info->npids); job->job.pidinfo->npgids = copy_int_list((int**)&(job->job.pidinfo->pgids), (int*)info->pgids, info->npgids); if (info->exdata != NULL) { job->job.pidinfo->exdata = makeString(info->exdata); } else { job->job.pidinfo->exdata = NULL; } return JOB_TRANSITION_APPLIED; } /* 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 (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; } xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_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. * * bug-1687: persist_submit_job_start_fact() carries `info` (the real * worker-reported pids/pgids/exdata) and a receipt-time start_time * alongside {jobid, state=RUN} -- see that function's doc comment and * wal_types.h's WAL_JOB_START payload note for the full rationale * (mirrors bug-1678's WAL_JOB_DISPATCH fix). Before this fix, * persist_submit_job_fact()'s bare {jobid, state} payload silently * dropped `info` entirely, so job.pidinfo replayed permanently NULL and * job.start_time was re-fabricated (time(NULL)) fresh at every replay * instead of carrying the true worker-reported time. Capturing * start_time HERE, once, at receipt time (rather than re-deriving * time(NULL) independently at each of apply_committed_entry's/ * apply_wal_entry's OWN apply-time moments) makes it the single * authoritative value both apply seams converge on. */ time_t start_time = time(NULL); (void)persist_submit_job_start_fact(&c->persist, job->job.jobid, JOB_STAT_RUN, (uint64_t)start_time, info); return 0; } int on_job_start(cluster_t* c, job_pidinfo_t* info) { job_t* job = NULL; int rc; char jobidStr[64]; /* bug-1588: see dispatch_a_job_()'s own comment on fmt_jobid(). */ fmt_jobid(jobidStr, sizeof(jobidStr), info->jobid); TRACE("%s jobid %s", __func__, jobidStr); lock(c->jobmgr, RTTI_LOCK_READ); rc = key_find(c->jobmgr, (void*)info->jobid, (void**)&job); if (job != NULL) { reference(job); } unlock(c->jobmgr); if ((rc <= 0) || (job == NULL)) { ERROR("can not find job %ld", info->jobid); return -errno; } lock(job, RTTI_LOCK_WRITE); record_job_start_fact(c, job, info); unlock(job); dereference(job); return 0; } /* 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; 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) { 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) — 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 JOB_TRANSITION_NOOP_DUPLICATE; } /* update job information */ if (finish->exit_code == 0) { job->job.state = JOB_STAT_DONE; } else { job->job.state = JOB_STAT_EXIT; } job->job.finish_time = time(NULL); /* fill job run time/rusage information */ if (job->job.exit_info == NULL) { job->job.exit_info = (job_exitinfo_t*)calloc(1, sizeof(job_exitinfo_t)); } if (job->job.exit_info != NULL) { job->job.exit_info->exitcode = finish->exit_code; } job->job.run_time = job->job.finish_time - job->job.start_time; return JOB_TRANSITION_APPLIED; } /* 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 (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; } xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_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. * * bug-1689: persist_submit_job_finish_fact() carries a receipt-time * finish_time and the worker-reported exit_code alongside {jobid, * state} -- see that function's doc comment and wal_types.h's * WAL_JOB_FINISH payload note for the full rationale (mirrors * bug-1687's WAL_JOB_START fix). Before this fix, persist_submit_job_ * fact()'s bare {jobid, state} payload silently dropped both: job. * exit_info replayed permanently NULL and job.finish_time (and its * derived job.run_time) were re-fabricated (time(NULL)) fresh at every * apply instead of carrying the true receipt-time value. Capturing * finish_time HERE, once, at receipt time (rather than re-deriving * time(NULL) independently at each of apply_committed_entry's/ * apply_wal_entry's OWN apply-time moments) makes it the single * authoritative value both apply seams converge on. */ time_t finish_time = time(NULL); (void)persist_submit_job_finish_fact(&c->persist, job->job.jobid, (finish->exit_code == 0) ? JOB_STAT_DONE : JOB_STAT_EXIT, (uint64_t)finish_time, finish->exit_code); return 0; } int on_job_finish(cluster_t* c, xce_jobfinish_t* finish) { job_t* job = NULL; char jobidStr[64]; /* bug-1588: see dispatch_a_job_()'s own comment on fmt_jobid(). */ fmt_jobid(jobidStr, sizeof(jobidStr), finish->jobid); TRACE("%s job finish jobid %s exit code %d", __func__, jobidStr, finish->exit_code); lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)finish->jobid, (void**)&job); if (job != NULL) { reference(job); } unlock(c->jobmgr); if (job == NULL) { ERROR("can not find job %ld", finish->jobid); return -errno; } lock(job, RTTI_LOCK_WRITE); record_job_finish_fact(c, job, finish); unlock(job); dereference(job); return 0; } /* 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) { switch (failinfo->stage) { case JOB_EXEC_STAGE_Q_POST_EXEC: case JOB_EXEC_STAGE_POST_EXEC: if ((job->job.state & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) { WARN("job %ld got post exec fail notification but job's exit " "status has not been set", job->job.jobid); job->job.state |= JOB_STAT_EXIT; } job->job.state |= JOB_STAT_PERR; break; default: job->job.state = JOB_STAT_EXIT; break; } job->job.finish_time = time(NULL); 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, 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; } /* 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. * * task 6.1 round 2 fix (merge-staleness corruption, Req 11.3): the resultant * state depends on the CURRENT job state (PERR merges onto an existing * DONE/EXIT terminal state rather than resetting it — apply_job_fail_ * transition's switch below). Earlier this was precomputed HERE, at receipt * time, on a throwaway shallow copy of `job`, and the precomputed absolute * result was baked into the WAL entry. That is unsafe under deferred apply: * another fact for the same job (e.g. a FINISH) can still be pending apply * when this FAIL is received, so `job->job.state` at receipt time may not * be the state this FAIL's merge should actually be computed against — * baking a stale snapshot silently clobbers whatever the other fact applies * first (bug found in task 6.1 round 1 review). Fixed by carrying ONLY * `failinfo->stage` (via _pad's WAL_JOB_FAIL_STAGE_FLAG — see wal_types.h's * doc comment on wal_jobstate_t._pad) and deferring the actual merge * computation to apply time, against the job's real current state then * (apply_committed_entry's WAL_JOB_FAIL case, persist.c). */ static int record_job_fail_fact(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); /* Durable append only — no mutation of the real `job`, no AFTER_EVENT * firing here (Req 5.1, 5.3). `state` is unused for this apply-time- * recomputed shape (0 is written for forward-compatibility/diagnostics * only); the WAL_JOB_FAIL_STAGE_FLAG bit in `_pad` tells * apply_committed_entry() to route it to the real * apply_job_fail_transition() core instead of an absolute overwrite. */ (void)persist_submit_job_fact_ex(&c->persist, WAL_JOB_FAIL, job->job.jobid, 0, WAL_JOB_FAIL_STAGE_FLAG | failinfo->stage); 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 * 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 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 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; } /* 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_postdone_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; /* 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); apply_job_update_transition(job, state); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_UPDATE, c, job); return 0; } /* Guarded state-transition core for a postdone fact, extracted from * on_job_postdone's old inline mutation (raft-apply-decoupling task 6.1 * round 2, Req 3.3/7.1-7.3). Unlike a generic UPDATE, postdone's formula * (conditionally backstop EXIT, then OR in JOB_STAT_PDONE) is a pure * function of the job's CURRENT state — so this core is meant to be invoked * AT APPLY TIME, against whatever state actually holds then, never against * a receipt-time snapshot (a receipt-time snapshot is exactly the merge- * staleness hazard this round fixes: another fact for the same job, e.g. a * FINISH, may still be in flight and apply AFTER receipt but BEFORE this * fact's own apply, under deferred apply). No-op when PDONE is already set * and no backstop is needed (idempotent re-apply, Req 3.4). Fires NO hook — * same convention as the other apply_job_*_transition cores; hook-firing * stays the caller's (apply_committed_entry's) responsibility. Caller must * hold `job` under RTTI_LOCK_WRITE. * * Known, previously-accepted residual (task 4.1, unchanged by this round): * finish_time is not part of the wal_jobstate_t wire format, so the old * defensive `job->job.finish_time = time(NULL)` backstop write has no * apply-time equivalent — a postdone arriving with no prior exit status * recorded still gets no defensive finish_time set anywhere. */ job_transition_result_t apply_job_postdone_transition(job_t* job) { int next = job->job.state; if ((next & (JOB_STAT_EXIT | JOB_STAT_DONE)) == 0) { WARN("job %ld got post exec fail notification but job's exit " "status has not been set", job->job.jobid); next |= JOB_STAT_EXIT; } next |= JOB_STAT_PDONE; if (next == job->job.state) { INFO("job %ld: duplicate postdone — no-op", job->job.jobid); return JOB_TRANSITION_NOOP_DUPLICATE; } job->job.state = next; return JOB_TRANSITION_APPLIED; } /* raft-apply-decoupling task 4.1/6.1 (Req 5.1, 3.3, 7.1-7.3): durable- * append-only sibling of the old inline postdone mutation, used only by * on_job_postdone. Replaces record_job_update_fact (removed in task 6.1 * round 2 — this was its only caller): postdone's mutation is a pure * function of current state (see apply_job_postdone_transition above), so * nothing needs to be computed or peeked at receipt time — the WAL entry * carries no state value at all, just a marker (_pad == * WAL_JOB_UPDATE_POSTDONE_MARK) telling apply_committed_entry() to re- * derive the merge fresh, against whatever the job's real current state is * when this entry actually applies (fixes the same class of merge- * staleness corruption record_job_fail_fact's fix addresses above). */ static int record_job_postdone_fact(cluster_t* c, job_t* job) { 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_ex(&c->persist, WAL_JOB_UPDATE, job->job.jobid, 0, WAL_JOB_UPDATE_POSTDONE_MARK); return 0; } int on_job_fail(cluster_t* c, job_failinfo_t* failinfo) { job_t* job = NULL; char jobidStr[64]; /* bug-1588: see dispatch_a_job_()'s own comment on fmt_jobid(). */ fmt_jobid(jobidStr, sizeof(jobidStr), failinfo->jobid); TRACE("%s job fail jobid %s code %d", __func__, jobidStr, failinfo->code); lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)failinfo->jobid, (void**)&job); if (job != NULL) { reference(job); } unlock(c->jobmgr); if (job == NULL) { ERROR("can not find job %ld", failinfo->jobid); return -errno; } lock(job, RTTI_LOCK_WRITE); record_job_fail_fact(c, job, failinfo); unlock(job); dereference(job); return 0; } int on_job_stop(cluster_t* c, uint64_t jobid) { job_t* job = NULL; int state; char jobidStr[64]; /* bug-1588: see dispatch_a_job_()'s own comment on fmt_jobid(). */ fmt_jobid(jobidStr, sizeof(jobidStr), jobid); TRACE("%s job stop jobid %s", __func__, jobidStr); lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&job); if (job != NULL) { reference(job); } unlock(c->jobmgr); if (job == NULL) { ERROR("can not find job %ld", jobid); return -errno; } lock(job, RTTI_LOCK_WRITE); state = job->job.state; xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_STOP, c, job); state |= JOB_STAT_USUSP; job->job.state = state; xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_STOP, c, job); unlock(job); dereference(job); return 0; } int on_job_resume(cluster_t* c, uint64_t jobid) { job_t* job = NULL; int state; char jobidStr[64]; /* bug-1588: see dispatch_a_job_()'s own comment on fmt_jobid(). */ fmt_jobid(jobidStr, sizeof(jobidStr), jobid); TRACE("%s job resume jobid %s", __func__, jobidStr); lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&job); if (job != NULL) { reference(job); } unlock(c->jobmgr); if (job == NULL) { ERROR("can not find job %ld", jobid); return -errno; } lock(job, RTTI_LOCK_WRITE); state = job->job.state; xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_RESUME, c, job); state &= ~JOB_STAT_USUSP; job->job.state = state; xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_RESUME, c, job); unlock(job); dereference(job); return 0; } int on_job_postdone(cluster_t* c, endpoint_t* ep, uint64_t jobid) { job_t* job = NULL; char jobidStr[64]; /* bug-1588: see dispatch_a_job_()'s own comment on fmt_jobid(). */ fmt_jobid(jobidStr, sizeof(jobidStr), jobid); TRACE("%s job post done jobid %s", __func__, jobidStr); lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&job); if (job != NULL) { reference(job); } unlock(c->jobmgr); if (job == NULL) { ERROR("can not find job %ld", jobid); return -errno; } /* raft-apply-decoupling task 6.1 round 2 (Req 5.1, 11.3): the backstop- * EXIT-if-unset + OR-in-PDONE mutation formerly computed HERE, at * receipt time, is now deferred entirely to apply_job_postdone_ * transition() (called from apply_committed_entry() once this fact * commits) — see that function's doc comment for why baking it here * against a receipt-time snapshot was unsafe under deferred apply. */ lock(job, RTTI_LOCK_WRITE); record_job_postdone_fact(c, job); unlock(job); dereference(job); return 0; } int on_job_ckpt(cluster_t* c, endpoint_t* ep, xce_ckptinfo_t* ckptinfo) { uint64_t jobid; char* info = NULL; job_t* j = NULL; if (ckptinfo == NULL) { ERROR("%s ckptinfo should not be NULL", __func__); return -EINVAL; } jobid = ckptinfo->jobid; info = ckptinfo->ckptinfo; TRACE("%s job %ld ckptinfo %s", __func__, jobid, info); lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&j); if (j != NULL) { reference(j); } unlock(c->jobmgr); if (j == NULL) { ERROR("%s can not find job %ld", __func__, jobid); return -errno; } lock(j, RTTI_LOCK_WRITE); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_CKPT, c, j); if (j->job.ckptinfo != NULL) { dereference(j->job.ckptinfo); } j->job.ckptinfo = makeString(info); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_CKPT, c, j); unlock(j); dereference(j); return 0; } int do_jobcontrol_op(cluster_t* c, endpoint_t* ep, xce_request_t* request) { int rc = 0; xce_reply_t reply; init_reply_by_req(&reply, request); rc = job_ctrl(c, request->body.jobctrlmsg); /* Commit barrier (ha-commit-barrier-rollout Req 1.1-1.4): the success * reply is sent only after the operation's WAL entry is majority- * committed; barrier failures (-ETIMEDOUT, ...) become the error * reply. Errors from job_ctrl reply immediately — no barrier wait. * job_ctrl released all entity locks before returning, so blocking * here holds no lock. timeout 0 = configured sync_timeout_ms. */ if (rc == 0) { rc = persist_commit_barrier(&c->persist, 0); if (rc != 0) WARN("%s commit barrier failed, rc %d", __func__, rc); } reply.status = rc; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); return rc; } int update_job_rusages(cluster_t* c, endpoint_t* ep, xce_request_t* request) { xce_job_rusages_t* rusages = request->body.updatemsg->data.jobrusages; job_t* j = NULL; worker_t* w = NULL; char* host = NULL; TRACE("%s trying to update job(%ld) rusage", __func__, rusages->jobid); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_RUSAGE_UPDATE, c, rusages); lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)rusages->jobid, (void**)&j); if (j != NULL) { reference(j); } unlock(c->jobmgr); if (j == NULL) { PERROR("jobmgr_lockjob"); return -errno; } w = get_ep_worker(c, ep); if (w == NULL) { dereference(j); return errno; } host = makeString(w->host.name); if (host == NULL) { dereference(j); return -ENOMEM; } TRACE("%s update job (%ld) rusage for host %s", __func__, rusages->jobid, host); lock(j, RTTI_LOCK_WRITE); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_JOB_UPDATE, c, j); update_host_runusage(&(j->job.run_usage), host, rusages->count, rusages->usages); /* bug-1735: derive job-level max_mem/avg_mem/used_mem/cpu_time from the * per-host entries update_host_runusage() just updated -- replaces the * standing "update other fields of job rusage later" FIXME. */ aggregate_job_runusage(j->job.run_usage); update_pids(&(j->job.pidinfo), rusages->npids, rusages->pids); update_pgids(&(j->job.pidinfo), rusages->npgids, rusages->pgids); /* TEMP diagnostic (bug-1731 job-finish-notify-loss investigation). * DEBUG3-gated -- see bug-1731's own "re-add before next occurrence" * recommendation; the DEBUG3 log-retention bump (bug-1731 support, * main.c) exists so this survives a live burst without rotating out. */ TRACE("DIAG update_job_rusages: jobid=%ld state=0x%x (about to fire " "live JOB_UPDATE hook)", (long)j->job.jobid, (unsigned)j->job.state); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_UPDATE, c, j); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_RUSAGE_UPDATE, c, rusages); unlock(j); dereference(j); return 0; } static int set_preason(cluster_t* c, xce_pendreason_t* reasons) { job_t* j = NULL; xce_job_t* job = NULL; char jobidStr[64]; /* bug-1588: see dispatch_a_job_()'s own comment on fmt_jobid(). */ fmt_jobid(jobidStr, sizeof(jobidStr), reasons->jobid); TRACE("%s jobid %s", __func__, jobidStr); lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)reasons->jobid, (void**)&j); if (j != NULL) { reference(j); } unlock(c->jobmgr); if (j == NULL) { PERROR("jobmgr_lockjob"); return -ENOENT; } lock(j, RTTI_LOCK_WRITE); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_SETPREASON, c, j); job = &(j->job); if (job->pend_reasons != NULL) { erase_job_preason(job->pend_reasons); FREEUP(job->pend_reasons); } MOVE_PTR(job->pend_reasons, reasons->reasons); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_SETPREASON, c, j); unlock(j); dereference(j); return 0; } int do_set_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request) { int i; xce_preason_set_t* reasons = request->body.preasonmsg; xce_reply_t reply; init_reply_by_req(&reply, request); reply.status = 0; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); for (i = 0; i < reasons->nreasons; i++) { int rc = 0; if (reasons->reasons[i] == NULL) { continue; } rc = set_preason(c, reasons->reasons[i]); if (rc < 0) { PERROR("set_preason"); } } return 0; } int do_clr_preason(cluster_t* c, endpoint_t* ep, xce_request_t* request) { int i; xce_reply_t reply; xce_clear_reason_t* clreason = request->body.clrpreasonmsg; init_reply_by_req(&reply, request); reply.status = 0; reply.data_which = RequestReply_data_nodata; xce_reply(ep, c->myhost, &reply); for (i = 0; i < clreason->njobs; i++) { uint64_t jobid = clreason->jobs[i]; job_t* j; xce_job_t* job; lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)jobid, (void**)&j); if (j != NULL) { reference(j); } unlock(c->jobmgr); if (j == NULL) { PERROR("jobmgr_lockjob"); continue; } lock(j, RTTI_LOCK_WRITE); xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_BEFORE_EVENT, HOOK_DETAIL_CLRPREASON, c, j); job = &(j->job); if (job->pend_reasons != NULL) { erase_job_preason(job->pend_reasons); FREEUP(job->pend_reasons); } xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_CLRPREASON, c, j); unlock(j); dereference(j); } return 0; } /* ---- fetchfile pending-request infrastructure ---- */ typedef struct fetchfile_pending { uint64_t jobid; worker_t* worker; /* bug-1693: which worker this request was * proxied to -- lets a disconnect be * matched to the pending waiter(s) it * affects; see * fetchfile_signal_worker_disconnect(). */ pthread_mutex_t mu; pthread_cond_t cv; int done; int status; xce_fetchfile_reply_t* reply; struct fetchfile_pending* next; } fetchfile_pending_t; static pthread_mutex_t g_fetchfile_mu = PTHREAD_MUTEX_INITIALIZER; static fetchfile_pending_t* g_fetchfile_list = NULL; static void fetchfile_register(fetchfile_pending_t* p) { pthread_mutex_lock(&g_fetchfile_mu); p->next = g_fetchfile_list; g_fetchfile_list = p; pthread_mutex_unlock(&g_fetchfile_mu); } static void fetchfile_unregister(fetchfile_pending_t* p) { fetchfile_pending_t** pp; pthread_mutex_lock(&g_fetchfile_mu); for (pp = &g_fetchfile_list; *pp != NULL; pp = &(*pp)->next) { if (*pp == p) { *pp = p->next; break; } } pthread_mutex_unlock(&g_fetchfile_mu); } fetchfile_pending_t* fetchfile_find(uint64_t jobid) { fetchfile_pending_t* p; pthread_mutex_lock(&g_fetchfile_mu); for (p = g_fetchfile_list; p != NULL; p = p->next) { if (p->jobid == jobid) break; } pthread_mutex_unlock(&g_fetchfile_mu); return p; } /* Signal a pending fetchfile request with the worker's reply. */ void fetchfile_signal(uint64_t jobid, int status, xce_fetchfile_reply_t* reply) { fetchfile_pending_t* p = fetchfile_find(jobid); if (p == NULL) return; pthread_mutex_lock(&p->mu); p->status = status; p->reply = reply; p->done = 1; pthread_cond_signal(&p->cv); pthread_mutex_unlock(&p->mu); } /* bug-1693 (Decision Log: "No client reply-wait timeout as a fix for a hung * request -- fail loud at the real transport/source instead of masking with * a deadline backstop"). Before this fix, handle_worker_leave() (worker.c) * only ever nulled w->ep on a worker disconnect -- it never woke a pending * on_fetchfile() waiter proxied to that worker, so a mid-wait disconnect * could only ever be observed at on_fetchfile's own 10s deadline below. * Called from handle_worker_leave() immediately on disconnect. Unlike * fetchfile_signal() (keyed by jobid, at most one match), more than one * fetchfile request can legitimately be in flight against the SAME worker * at once, so this walks the whole list and signals every match -- a no-op * if none are pending against `w`. Deliberately takes only g_fetchfile_mu + * each matched waiter's own mu, never any worker_t lock, so callers do not * need to hold (or avoid holding) `w`'s lock around this call. */ void fetchfile_signal_worker_disconnect(worker_t* w) { fetchfile_pending_t* p; pthread_mutex_lock(&g_fetchfile_mu); for (p = g_fetchfile_list; p != NULL; p = p->next) { if (p->worker != w) continue; pthread_mutex_lock(&p->mu); p->status = -ECONNRESET; p->reply = NULL; p->done = 1; pthread_cond_signal(&p->cv); pthread_mutex_unlock(&p->mu); } pthread_mutex_unlock(&g_fetchfile_mu); } #ifdef XCE_TESTING /* Test seam (bug-1693): exercises fetchfile_signal_worker_disconnect()'s * demux against a real fetchfile_pending_t wait, without needing a full * on_fetchfile()/real c->jobmgr+c->workermgr fixture (job/worker lookup is * out of this fix's scope -- only the disconnect-to-waiter signal path is * under test). Registers a bare pending entry keyed to `w`, blocks with a * deadline exactly like on_fetchfile's own wait loop below, and reports the * outcome via *out_status. Returns 1 if woken by a signal, 0 if the * deadline lapsed first (mirrors pending.done). */ int fetchfile_wait_for_test(worker_t* w, long timeout_ms, int* out_status) { fetchfile_pending_t pending = { .jobid = 0, .worker = w, .mu = PTHREAD_MUTEX_INITIALIZER, .cv = PTHREAD_COND_INITIALIZER, .done = 0, .status = -ETIMEDOUT, .reply = NULL, .next = NULL }; struct timespec deadline; clock_gettime(CLOCK_REALTIME, &deadline); deadline.tv_sec += timeout_ms / 1000; deadline.tv_nsec += (timeout_ms % 1000) * 1000000L; if (deadline.tv_nsec >= 1000000000L) { deadline.tv_sec += 1; deadline.tv_nsec -= 1000000000L; } fetchfile_register(&pending); pthread_mutex_lock(&pending.mu); while (!pending.done) { int wrc = pthread_cond_timedwait(&pending.cv, &pending.mu, &deadline); if (wrc == ETIMEDOUT) break; } pthread_mutex_unlock(&pending.mu); fetchfile_unregister(&pending); *out_status = pending.status; return pending.done; } #endif /* ---- on_fetchfile: proxy fetchfile from apiserver to worker ---- */ int on_fetchfile(cluster_t* c, endpoint_t* caller_ep, xce_request_t* request) { xce_fetchfile_t* msg = request->body.fetchfilemsg; xce_reply_t reply; job_t* j = NULL; worker_t* w = NULL; int rc = 0; char* exechost = NULL; struct timespec deadline; fetchfile_pending_t pending = { .mu = PTHREAD_MUTEX_INITIALIZER, .cv = PTHREAD_COND_INITIALIZER, .done = 0, .status = -ETIMEDOUT, .reply = NULL, .next = NULL }; init_reply_by_req(&reply, request); if (msg == NULL || msg->filepath == NULL || msg->filepath[0] == '\0') { reply.status = -EINVAL; reply.data_which = RequestReply_data_nodata; xce_reply(caller_ep, c->myhost, &reply); return -EINVAL; } pending.jobid = msg->jobid; /* Look up the job */ lock(c->jobmgr, RTTI_LOCK_READ); key_find(c->jobmgr, (void*)msg->jobid, (void**)&j); if (j != NULL) reference(j); unlock(c->jobmgr); if (j == NULL) { reply.status = -ENOENT; reply.data_which = RequestReply_data_nodata; xce_reply(caller_ep, c->myhost, &reply); return -ENOENT; } /* Get the first exec-host from job allocs */ lock(j, RTTI_LOCK_READ); if (j->job.allocs != NULL && j->job.allocs->nhosts > 0 && j->job.allocs->allocs != NULL && j->job.allocs->allocs[0] != NULL && j->job.allocs->allocs[0]->host != NULL) { exechost = strdup(j->job.allocs->allocs[0]->host); } unlock(j); dereference(j); if (exechost == NULL) { reply.status = -ENOENT; reply.data_which = RequestReply_data_nodata; xce_reply(caller_ep, c->myhost, &reply); return -ENOENT; } /* Find the worker */ lock(c->workermgr, RTTI_LOCK_READ); rc = key_find(c->workermgr, (void*)exechost, (void**)&w); if (w != NULL) reference(w); unlock(c->workermgr); free(exechost); if (w == NULL) { reply.status = -ENOENT; reply.data_which = RequestReply_data_nodata; xce_reply(caller_ep, c->myhost, &reply); return -ENOENT; } lock(w, RTTI_LOCK_READ); if (w->ep == NULL) { unlock(w); dereference(w); reply.status = -ENOENT; reply.data_which = RequestReply_data_nodata; xce_reply(caller_ep, c->myhost, &reply); return -ENOENT; } /* bug-1693: record which worker this request is proxied to (still * holding w's read lock, so w->ep/identity are stable here) so a later * disconnect can find and wake this waiter -- see * fetchfile_signal_worker_disconnect()'s doc comment above. */ pending.worker = w; /* Register pending before sending to avoid a race with the reply */ fetchfile_register(&pending); rc = xce_fetchfile_req(w->ep, msg); unlock(w); dereference(w); if (rc < 0) { fetchfile_unregister(&pending); reply.status = rc; reply.data_which = RequestReply_data_nodata; xce_reply(caller_ep, c->myhost, &reply); return rc; } /* Wait up to 10 seconds for the worker to respond */ clock_gettime(CLOCK_REALTIME, &deadline); deadline.tv_sec += 10; pthread_mutex_lock(&pending.mu); while (!pending.done) { rc = pthread_cond_timedwait(&pending.cv, &pending.mu, &deadline); if (rc == ETIMEDOUT) break; } pthread_mutex_unlock(&pending.mu); fetchfile_unregister(&pending); /* Relay the reply back to the apiserver */ if (pending.done && pending.status >= 0 && pending.reply != NULL) { reply.status = 0; reply.data_which = RequestReply_data_fetchfilereply; reply.data.fetchfilereply = pending.reply; } else { reply.status = pending.done ? pending.status : -ETIMEDOUT; reply.data_which = RequestReply_data_nodata; FREEUP(pending.reply); } xce_reply(caller_ep, c->myhost, &reply); return 0; } /* job_is_terminal and job_should_retain live in persist.c — * they only depend on xce_job_t fields and persist_config_t, * and must be accessible from the unit test binary which links persist.c. */