myxce/src/controller/job.c

#include <assert.h>
#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <string.h>
#include <time.h>

#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. */
int restore_job_dispatch_state(cluster_t* c, uint64_t jobid, uint32_t state,
                               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 (alloc != NULL) {
        if (j->job.allocs != NULL)
            free_JobAlloc_ptr(&(j->job.allocs));
        j->job.allocs = alloc;
    }
    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;

        /* 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. */
        (void)persist_submit_job_dispatch_fact(&c->persist, alloc->jobid,
                                               JOB_STAT_RUN, j->job.allocs);

        unlock(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. */
    (void)persist_submit_job_fact(&c->persist, WAL_JOB_START, job->job.jobid,
                                  JOB_STAT_RUN);

    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. */
    (void)persist_submit_job_fact(&c->persist, WAL_JOB_FINISH, job->job.jobid,
                                  (finish->exit_code == 0) ? JOB_STAT_DONE
                                                            : JOB_STAT_EXIT);

    return 0;
}

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);

    update_pids(&(j->job.pidinfo), rusages->npids, rusages->pids);

    update_pgids(&(j->job.pidinfo), rusages->npgids, rusages->pgids);

    /* FIXME
     * update other fields of job rusage later
     */
    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;
    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);
}

/* ---- 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;
    }

    /* 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. */