myxce/src/controller/resync.c

/* resync.c -- worker-rejoin-resync task 3.1 (Req 1.1-1.3, 2.1, 2.2, 5.3,
 * 6.1, 6.3): per-worker job-state reconciliation round lifecycle.
 *
 * Owns exactly what design.md's "Controller: resync.c" section assigns it:
 * arm a round (query-set construction + round token + deadline + a
 * non-blocking delaymgr one-shot timeout), resolve a round (on timeout or,
 * later, on normal completion -- task 3.2), and a round-token staleness
 * check (Req 6.3) later callers use to discard a report that arrives for an
 * already-discarded/superseded round.
 *
 * Explicitly NOT this file's job (later tasks): sending the query/report
 * over the wire (task 6.1's worker.c/on_worker_join wiring), and mapping a
 * reported job-state entry to the correct job.c fact-recording call (task
 * 3.2's report-apply glue). Every function here is written to be called by
 * that later wiring, not to include it.
 */
#include <errno.h>
#include <rtti.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#include "cluster.h"
#include "controller.capnp.h" /* JobStateStatus_* -- worker-rejoin-resync
                                * task 3.2 */
#include "delay.h"
#include "persist.h" /* persist_notify_job_updated_already_persisted --
                       * worker-rejoin-resync task 3.2 */
#include "resync.h"
#include "utils.h"
#include "xce/types.h"

/* ── Round token ─────────────────────────────────────────────────────────
 *
 * Req 6.1/6.3: a fresh round must be distinguishable from any prior round
 * for the SAME worker (a disconnect-then-rejoin re-arming the round, or a
 * timeout racing a late report). A single process-wide monotonic counter,
 * shared across every worker's rounds, trivially satisfies this: every arm
 * anywhere gets a value strictly greater than every arm before it, so in
 * particular greater than any prior round for the specific worker being
 * armed right now. This is simpler and collision-free versus a wall-clock
 * timestamp (which can repeat within the same millisecond under load, or
 * even go backward across an NTP step) or a per-worker counter (extra state
 * to carry and initialize per worker_t). Starts at 1 so 0 stays available
 * as an unambiguous "no token" sentinel if ever needed. */
static _Atomic uint64_t g_resync_next_round_token = 1;

static uint64_t resync_next_round_token(void)
{
    return atomic_fetch_add_explicit(&g_resync_next_round_token, 1,
                                     memory_order_relaxed);
}

/* ── Timeout / deadline ──────────────────────────────────────────────────
 *
 * Req 5.3's timeout value is ClusterConfig.workerResyncTimeoutMs
 * (worker_resync_timeout_ms on the wire-decoded cluster_config_t,
 * clusterconfig.h), reached via c->myconf->cluster_config -- the same
 * accessor chain already used for the sibling shared-config field
 * heartbeat_interval (cluster.c:1347/2270's
 * c->myconf->cluster_config->heartbeat_interval). No xceadmin/schema
 * default is defined for this field (src/configs/schema/xce.schema.json
 * declares it with no "default" key); the fixture value used throughout
 * xceadmin's own tests (src/xceadmin/tests/test_mapping.py,
 * fixtures/full/configs/xce.yaml) is 15000ms, adopted here as the fallback
 * for when a live cluster_config isn't available at all (myconf/
 * cluster_config unset or the field left at its zero-value) -- e.g. a
 * minimal unit-test cluster_t fixture that never populates myconf, mirroring
 * myconf.c's own SET_DEFAULT_IF_UNSET convention for heartbeat_interval. */
#define RESYNC_DEFAULT_TIMEOUT_MS 15000

static uint64_t resync_timeout_ms(cluster_t* c)
{
    if (c != NULL && c->myconf != NULL && c->myconf->cluster_config != NULL
        && c->myconf->cluster_config->worker_resync_timeout_ms > 0)
        return (uint64_t)c->myconf->cluster_config->worker_resync_timeout_ms;

    return RESYNC_DEFAULT_TIMEOUT_MS;
}

/* Same CLOCK_MONOTONIC millisecond representation resync_state_t.deadline
 * documents (xce/types.h) and delay.h's own delay_task_t.nextrun_ms uses
 * (delay.c's monotonic_ms(), duplicated here rather than exposed by
 * delay.h/delay.c -- it is a two-line static helper, not worth a new public
 * API for). */
static uint64_t resync_monotonic_ms(void)
{
    struct timespec now;

    clock_gettime(CLOCK_MONOTONIC, &now);
    return (uint64_t)now.tv_sec * 1000ULL + (uint64_t)now.tv_nsec / 1000000ULL;
}

/* ── Query-set construction (Req 1.1, 2.1, 2.2) ─────────────────────────── */

/* Mirrors job.c's own private JOB_STATE_TERMINAL_BITS/job_state_is_terminal
 * (job.c:41-50): a job is terminal once it carries a DONE or EXIT bit.
 * job.c's predicate is `static inline` and not exported via cluster.h, and
 * this task's boundary is explicitly "read job.c for reference, don't
 * modify it" -- so this is an independent definition built from the SAME
 * public bit constants (xce/types.h's JOB_STAT_DONE/JOB_STAT_EXIT), not a
 * duplicate of a private symbol and not a change to job.c to export one. */
static inline int resync_job_is_terminal(unsigned int state)
{
    return (state & (JOB_STAT_DONE | JOB_STAT_EXIT)) != 0;
}

/* True iff `j`'s head/primary assigned host matches `hostname`. A job's
 * host assignment lives ENTIRELY in job->job.allocs->allocs[0]->host --
 * confirmed by reading job.c's own existing readers of this same
 * information (job.c's mark_job_ctrl-family job-control routing at
 * job.c:1922-1936; the exechost lookup at job.c:3766-3771) and by
 * record_job_reconcile_assign_fact's own doc comment (job.c:3031-3045),
 * which independently confirms neither job_t nor xce_job_t carries any
 * OTHER field representing a job's assigned host. allocs[0] specifically
 * is the job's HEAD host, matching requirements.md's Boundary Context
 * ("Multi-host jobs... the Controller dispatches and tracks a multi-host
 * job as a single record owned by that job's head/primary host"). Caller
 * holds `j`'s read lock. */
static int resync_job_assigned_to_host(job_t* j, const char* hostname)
{
    xce_job_alloc_t* alloc = j->job.allocs;

    if (alloc == NULL || alloc->nhosts <= 0 || alloc->allocs == NULL
        || alloc->allocs[0] == NULL || alloc->allocs[0]->host == NULL)
        return 0;

    return strcmp(alloc->allocs[0]->host, hostname) == 0;
}

/* Req 1.1, 2.1, 2.2: builds the round's query-job-ID set -- every
 * non-terminal job in c->jobmgr currently assigned to w's host. Locking
 * mirrors persist.c's collect_job_frames() nested discipline exactly: the
 * jobmgr table's own RTTI_LOCK_READ is held for the entire walk (excludes
 * concurrent insert/delete), and each candidate job's OWN RTTI_LOCK_READ is
 * additionally, nestedly, taken just around reading its state/allocs --
 * already the proven-safe pattern in this codebase for a point-in-time-
 * consistent jobmgr walk. Independent of `w`'s own lock (a different lock
 * object entirely), so this never conflicts with the write lock the caller
 * of resync_arm_round already holds on `w`.
 *
 * On success (return 0), *out_ids is a malloc'd uint64_t[*out_n] the caller
 * owns (free() when done); *out_n may be 0 with *out_ids left NULL -- Req
 * 1.2's "no non-terminal jobs for this host" case is not an error. On
 * failure (negative errno), *out_ids and *out_n are left NULL/0 and nothing
 * remains allocated. */
static int resync_collect_query_jobids(cluster_t* c, worker_t* w,
                                       uint64_t** out_ids, int* out_n)
{
    iterator_t it;
    int        rc;
    uint64_t*  ids = NULL;
    int        n   = 0;
    int        cap = 0;

    *out_ids = NULL;
    *out_n   = 0;

    if (c == NULL || c->jobmgr == NULL || w == NULL)
        return -EINVAL;

    lock(c->jobmgr, RTTI_LOCK_READ);

    rc = iter_make(&it, c->jobmgr);
    if (rc < 0) {
        unlock(c->jobmgr);
        return 0; /* empty table -- zero matches, not an error (mirrors
                   * collect_job_frames'/collect_jobs' own empty-table
                   * handling) */
    }

    for (;;) {
        job_t*   j = NULL;
        int      nrc;
        int      match;
        uint64_t jobid;

        nrc = iter_next(&it, (void**)&j);
        if (nrc < 0 || nrc == RTTI_ITER_NO_DATA)
            break;
        if (j == NULL)
            continue;

        lock(j, RTTI_LOCK_READ);
        match = !resync_job_is_terminal(j->job.state)
                && resync_job_assigned_to_host(j, w->host.name);
        jobid = j->job.jobid;
        unlock(j);

        if (!match)
            continue;

        if (n == cap) {
            int       newcap = (cap == 0) ? 8 : (cap * 2);
            uint64_t* grown
                = (uint64_t*)realloc(ids, (size_t)newcap * sizeof(uint64_t));
            if (grown == NULL) {
                free(ids);
                iter_destroy(&it);
                unlock(c->jobmgr);
                *out_ids = NULL;
                *out_n   = 0;
                return -ENOMEM;
            }
            ids = grown;
            cap = newcap;
        }
        ids[n++] = jobid;
    }

    iter_destroy(&it);
    unlock(c->jobmgr);

    *out_ids = ids;
    *out_n   = n;
    return 0;
}

/* ── Timeout trampoline (Req 5.3) ────────────────────────────────────────
 *
 * delay.c/delay.h expose no "cancel a queued task" primitive (delay_task_ms
 * only ever enqueues; delay_worker's own loop only ever fires-then-frees or
 * re-arms). So a round that resolves normally (task 3.2, later) BEFORE its
 * timeout fires cannot make the already-queued delaymgr task disappear --
 * it will still fire, eventually, on schedule. Safety against a stale fire
 * acting on an already-cleared/reused resync_state_t therefore comes from
 * the round token, not from dequeuing: the closure below captures the
 * round token the timeout was armed FOR, and resync_on_timeout only acts if
 * resync_round_is_current() still reports that exact token as current for
 * `w` at fire time. A round that already resolved (normally, or via a
 * newer resync_arm_round superseding it) makes the old token stale, so the
 * late fire becomes a pure no-op -- never a double-clear, never touching a
 * `resync_state_t` that was already freed and possibly replaced. */
typedef struct {
    cluster_t* c;
    worker_t*  w;
    uint64_t   round_token;
    uint64_t   timeout_ms;
} resync_timeout_ctx_t;

/* Runs on the delaymgr timer thread (c->delaymgr's runner_fn, cluster.c's
 * run_delay_tasks, calls this directly and synchronously). Per the
 * async-I/O thread-role contract (ha-async-io.md's Thread-role contract
 * DESIGN INVARIANT): delaymgr must fire timers and trigger work, then
 * return fast -- no connect(), no file I/O, no unbounded wait. Everything
 * below is bounded, in-memory bookkeeping under `w`'s own mutex (a status
 * bit clear, freeing one small heap struct) -- never a blocking call, so
 * this satisfies that contract by construction, not by inspection alone.
 *
 * worker_t* is captured directly (not re-looked-up via c->workermgr):
 * confirmed by reading cluster.c that no code path ever key_del()s an entry
 * from c->workermgr once created (worker_obj_killer, its kvmgr datakiller,
 * is a documented no-op) -- a worker_t, once created, lives for the
 * process's lifetime, the same assumption ha.c's own delaymgr trampolines
 * (e.g. ha_phase2_retry_delaymgr_trampoline capturing cluster_t* c) already
 * make about their own captured pointers. */
static int resync_on_timeout(void* arg)
{
    resync_timeout_ctx_t* ctx = (resync_timeout_ctx_t*)arg;

    if (ctx == NULL)
        return -EINVAL;

    lock(ctx->w, RTTI_LOCK_WRITE);
    if (resync_round_is_current(ctx->w, ctx->round_token)) {
        /* Req 5.3: "accepted-as-answered, timed out, or left for the next
         * rejoin to retry" -- this implementation takes the literal
         * "timed out" branch: resolve the round now so the worker is never
         * withheld from dispatch indefinitely (Req 5.1/5.2). A report that
         * later arrives for this now-resolved round is naturally discarded
         * by resync_round_is_current's own stale-token check (Req 6.3)
         * wherever a later caller (task 3.2) applies it -- there is no
         * separate "reattempted on a subsequent rejoin" action to take
         * here; that is what a fresh resync_arm_round on the next join
         * already provides. */
        WARN("resync: worker %s reconciliation round %lu timed out after "
             "%lums -- resolving so dispatch is not withheld indefinitely "
             "(Req 5.3)",
             (ctx->w->host.name != NULL) ? ctx->w->host.name : "?",
             (unsigned long)ctx->round_token,
             (unsigned long)ctx->timeout_ms);
        resync_resolve_round(ctx->c, ctx->w);
    }
    /* else: stale fire -- the round this timeout was armed for already
     * resolved (normally) or was superseded by a newer resync_arm_round
     * before the timer got here. No-op by construction (see the comment
     * above resync_timeout_ctx_t). */
    unlock(ctx->w);

    free(ctx);
    return 0;
}

/* ── Public API ──────────────────────────────────────────────────────── */

int resync_arm_round(cluster_t* c, worker_t* w)
{
    uint64_t*             ids = NULL;
    int                    n   = 0;
    int                    rc;
    resync_state_t*        rs;
    resync_timeout_ctx_t*  ctx;
    uint64_t               timeout_ms;

    if (c == NULL || w == NULL)
        return -EINVAL;

    /* Req 6.1: an already-outstanding round for this worker (this rejoin
     * interrupts one that never resolved) is discarded first, so the fresh
     * round's token always supersedes it -- never two live resync_state_t
     * generations for the same worker_t at once. */
    if (w->resync != NULL)
        resync_resolve_round(c, w);

    rc = resync_collect_query_jobids(c, w, &ids, &n);
    if (rc < 0) {
        ERROR("resync_arm_round: worker %s: failed to build reconciliation "
             "query set: %d -- no round armed (the next rejoin will "
             "attempt again)",
             (w->host.name != NULL) ? w->host.name : "?", rc);
        return rc;
    }

    if (n == 0) {
        /* Req 1.2: no non-terminal jobs for this host -- documented no-op:
         * no round armed, no status bit set, no timer scheduled. */
        free(ids);
        return 0;
    }

    rs = (resync_state_t*)calloc(1, sizeof(*rs));
    if (rs == NULL) {
        free(ids);
        return -ENOMEM;
    }

    rs->round_token      = resync_next_round_token();
    rs->n_queried_jobids = n;
    rs->queried_jobids   = ids;

    timeout_ms   = resync_timeout_ms(c);
    rs->deadline = resync_monotonic_ms() + timeout_ms;

    w->resync = rs;
    w->host.status |= XCE_WORKER_RECONCILING;

    ctx = (resync_timeout_ctx_t*)calloc(1, sizeof(*ctx));
    if (ctx == NULL) {
        /* The round itself is still validly armed (query set, token,
         * deadline, status bit are all already set above) -- only the
         * BOUNDED-TIMEOUT half of Req 5.3 is unavailable this time. Fail
         * loud and stay alive (steering: never fabricate partial success
         * silently) rather than unwind an otherwise-good round over an
         * allocation failure for a small bookkeeping struct; a later
         * report (task 3.2) still resolves it normally. */
        ERROR("resync_arm_round: worker %s round %lu: timeout context "
             "alloc failed -- round armed WITHOUT a bounded timeout (Req "
             "5.3 degraded: resolves normally on report, not via timeout "
             "this time)",
             (w->host.name != NULL) ? w->host.name : "?",
             (unsigned long)rs->round_token);
        return n;
    }
    ctx->c           = c;
    ctx->w           = w;
    ctx->round_token = rs->round_token;
    ctx->timeout_ms  = timeout_ms;

    if (!c->delaymgr.running) {
        /* Not-yet-started/test-fixture delaymgr -- mirrors ha.c's own
         * `!c->delaymgr.running` guard (e.g. schedule_upgrade_
         * materialization_retry, ha_schedule_phase2_retry): the round
         * stays armed; a test drives resync_on_timeout directly instead of
         * waiting on a real timer thread that was never started. */
        free(ctx);
        return n;
    }

    rc = delay_task_ms(&c->delaymgr, (int)timeout_ms, 1, resync_on_timeout,
                       ctx);
    if (rc < 0) {
        WARN("resync_arm_round: worker %s round %lu: delay_task_ms failed: "
             "%d -- round armed without a bounded timeout",
             (w->host.name != NULL) ? w->host.name : "?",
             (unsigned long)rs->round_token, rc);
        free(ctx);
    }

    return n;
}

void resync_resolve_round(cluster_t* c, worker_t* w)
{
    (void)c; /* accepted to match this file's other functions' shape and
              * what a later caller (task 3.2) already has in hand
              * alongside `w` -- this function's own body has nothing to do
              * with `c` today. */

    if (w == NULL || w->resync == NULL)
        return;

    w->host.status &= ~XCE_WORKER_RECONCILING;

    free(w->resync->queried_jobids);
    free(w->resync);
    w->resync = NULL;
}

int resync_round_is_current(worker_t* w, uint64_t token)
{
    if (w == NULL || w->resync == NULL)
        return 0;

    return w->resync->round_token == token;
}

/* ── Report application (task 3.2, Req 4.1-4.5, 6.1-6.3) ─────────────────
 *
 * Maps each JobStateEntry in a worker's job-state report to the correct
 * idempotent job.c fact-recording call, then resolves the round once every
 * queried job has a reconciled outcome. See resync.h's doc comment on
 * resync_apply_report() for the full per-outcome routing contract, and
 * design.md's "System Flows" sequence diagram (the `loop each reported
 * entry` block) for the design this implements.
 */

/* Req 6.2: removes `jobid` from w->resync's outstanding query set, if
 * present -- swap-with-last-and-shrink (the array is small, bounded by one
 * host's non-terminal job count per resync_collect_query_jobids' own
 * reasoning, so O(n) removal is not a concern). A no-op when w->resync is
 * NULL or `jobid` is not present -- safe to call unconditionally and
 * repeatedly (an already-cleared id, or one that was never queried, e.g. an
 * unqueried extra). Caller holds w's write lock. */
static void resync_clear_outstanding(worker_t* w, uint64_t jobid)
{
    resync_state_t* rs = w->resync;
    int              i;

    if (rs == NULL)
        return;

    for (i = 0; i < rs->n_queried_jobids; i++) {
        if (rs->queried_jobids[i] == jobid) {
            rs->queried_jobids[i]
                = rs->queried_jobids[rs->n_queried_jobids - 1];
            rs->n_queried_jobids--;
            return;
        }
    }
}

/* Req 5.1/5.2: resolves w's round if its outstanding set has just become
 * empty. Caller holds w's write lock and must have already confirmed the
 * round is still current for the token in question (see
 * resync_settle_entry below) -- this function itself only re-checks
 * w->resync != NULL (resync_resolve_round's own precondition), not token
 * identity, since the caller's own check under the SAME uninterrupted lock
 * hold already establishes that. */
static void resync_maybe_resolve(cluster_t* c, worker_t* w)
{
    if (w->resync != NULL && w->resync->n_queried_jobids == 0)
        resync_resolve_round(c, w);
}

/* Req 4.1, 5.1, 5.2, 6.1-6.3: clears `jobid` from w's outstanding set for
 * round `token` and, if the set has just become empty, resolves the round
 * -- both performed under a SINGLE w write-lock acquisition, both gated on
 * `token` still being w's CURRENT round. The gate matters: a concurrent Req
 * 5.3 timeout fire, or a fresh resync_arm_round superseding this round on
 * another thread, can happen between this report-apply loop's per-entry
 * fact application (which holds no lock on `w`) and this call -- without
 * re-validating here, a stale round's late bookkeeping could otherwise
 * mutate a DIFFERENT (newer) round's resync_state_t. This is also the exact
 * mechanism that makes re-applying an already-fully-resolved report a safe
 * no-op (Req 6.2): once resync_resolve_round() has run, w->resync is NULL,
 * so `token` no longer reports current and this becomes a pure no-op. */
static void resync_settle_entry(cluster_t* c, worker_t* w, uint64_t token,
                                uint64_t jobid)
{
    lock(w, RTTI_LOCK_WRITE);
    if (resync_round_is_current(w, token)) {
        resync_clear_outstanding(w, jobid);
        resync_maybe_resolve(c, w);
    }
    unlock(w);
}

/* Looks up `jobid` in c->jobmgr, reference()-ing it on success -- the same
 * key_find+reference idiom on_job_start/on_job_finish/on_job_fail/
 * dispatch_a_job_ (job.c) already use throughout this codebase. Returns
 * NULL if not found -- never treated as an error by this file's own callers
 * (see resync_apply_running_entry's doc comment for why a genuinely-
 * untracked jobid is a legitimate, if narrow, possibility). Caller must
 * dereference() the result when done, if non-NULL. */
static job_t* resync_lookup_job(cluster_t* c, uint64_t jobid)
{
    job_t* j = NULL;

    lock(c->jobmgr, RTTI_LOCK_READ);
    key_find(c->jobmgr, (void*)jobid, (void**)&j);
    if (j != NULL)
        reference(j);
    unlock(c->jobmgr);

    return j;
}

/* Req 4.1, 4.4, 4.7, 4.8: routes a "running" entry. A QUERIED running job's
 * own state already correctly reflects "running" -- no fact is recorded
 * (design.md's sequence diagram: "else still running -> clear this jobid
 * from outstanding set (no-op fact)"); the caller clears the outstanding
 * set for every queried entry uniformly (resync_apply_report below), so
 * this function has nothing further to do for that case.
 *
 * An UNQUERIED extra (Req 3.6) running job is a genuine THREE-WAY split, not
 * a two-way boolean (review round 2 regression fix, following round 1's own
 * review-caught fix -- the round-1 guard introduced a NEW Req 4.4 violation:
 * `!terminal && assigned-to-w` short-circuits to `false` whenever the job IS
 * terminal, falling through to the assign branch and unconditionally
 * clobbering a DONE/EXIT job back to RUN -- confirmed live: round 1's own
 * test seeded a TERMINAL j_extra and asserted it became RUN, certifying the
 * bug as "correct" instead of catching it):
 *
 * (a) TERMINAL (resync_job_is_terminal()) -> ALWAYS a no-op, regardless of
 *     host. Requirement 4.4 ("shall not overwrite that state with a less
 *     final or older report") is unconditional -- a terminal job is never
 *     assign-eligible, full stop. record_job_reconcile_assign_fact() has NO
 *     backward-transition guard of its own (its own doc comment: "no
 *     reject/no-op case" is a documented CALLER precondition, never
 *     re-checked internally) and unconditionally sets job->job.state =
 *     JOB_STAT_RUN -- calling it here would clobber DONE/EXIT back to RUN.
 * (b) Non-terminal AND job->job.allocs != NULL (already has SOME live
 *     non-terminal host assignment -- to THIS reporting worker's own host,
 *     or to a DIFFERENT one, it does not matter which) -> a no-op. It is
 *     already validly assigned; record_job_reconcile_assign_fact() would
 *     still be wrong to call even though its net STATE effect looks
 *     harmless (RUN -> RUN) -- it unconditionally resets job->job.start_time
 *     and re-appends a fresh WAL_JOB_DISPATCH fact, neither of which is
 *     warranted for a job that was never actually lost/unassigned.
 * (c) Non-terminal AND job->job.allocs == NULL (genuinely UNASSIGNED) ->
 *     Req 4.8's real trigger condition ("no existing non-terminal record...
 *     assigned elsewhere" -- there IS a non-terminal record, but it carries
 *     no assignment at all) -- reconciled via record_job_reconcile_assign_
 *     fact() (task 2.2), documented (and hard-asserted, job.c) as the
 *     running-only case.
 *
 * Concrete motivating scenario for (c) (the exact case design.md's own
 * "Documented scope decision" paragraph describes as this function's
 * target, and record_job_reconcile_assign_fact()'s core justification for
 * existing at all): a job the controller currently shows PEND (non-
 * terminal, allocs == NULL, unassigned to any host) because its
 * WAL_JOB_DISPATCH fact never reached durable majority commit before a
 * controller failover -- even though dispatch_a_job_() already sent
 * xce_startjob_req() to the worker and the worker really is running it (the
 * RPC fires BEFORE dispatch_a_job_()'s own eager mutation + WAL append, so
 * the two can diverge exactly at a failover boundary). Such a job correctly
 * reconciles via record_job_reconcile_assign_fact() (host now matches the
 * reporting worker, allocs stays NULL per Req 4.8's own documented scope
 * decision).
 *
 * A jobid with NO record AT ALL in the job table cannot be reconciled via
 * record_job_reconcile_assign_fact() either -- that function mutates an
 * EXISTING job_t, it does not construct one -- so this is logged and
 * skipped rather than fabricating a full job record from a bare entry;
 * every job a worker can legitimately run was, by this system's own
 * architecture, submitted through this controller at some point, so this
 * branch is expected to be unreachable in practice, not a routine path. */
static void resync_apply_running_entry(cluster_t* c, worker_t* w,
                                       xce_job_state_entry_t* entry)
{
    job_t* job;

    if (entry->queried)
        return;

    job = resync_lookup_job(c, entry->jobid);
    if (job == NULL) {
        WARN("resync: unqueried extra job %lu reported running by worker "
             "%s has no record at all in the job table -- cannot reconcile "
             "(no existing job_t to assign onto); skipped",
             (unsigned long)entry->jobid,
             (w->host.name != NULL) ? w->host.name : "?");
        return;
    }

    lock(job, RTTI_LOCK_WRITE);
    if (resync_job_is_terminal(job->job.state)) {
        /* (a) Terminal is always final -- Req 4.4. Never assign-eligible,
         * regardless of host. */
        unlock(job);
    } else if (job->job.allocs != NULL) {
        /* (b) Non-terminal AND already has SOME live non-terminal host
         * assignment -- a genuine no-op; do not reset start_time or
         * re-append a WAL_JOB_DISPATCH fact for a job that was never
         * actually lost/unassigned. */
        unlock(job);
    } else {
        /* (c) Non-terminal AND genuinely UNASSIGNED -- Req 4.8 fires. */
        int rc = record_job_reconcile_assign_fact(c, job, w, entry);
        unlock(job);
        if (rc == (int)JOB_TRANSITION_APPLIED)
            persist_notify_job_updated_already_persisted(c, job);
    }

    dereference(job);
}

/* Req 4.1-4.5, 4.7: routes a "finished" entry (success or failure) through
 * the SAME existing, deferred-pattern fact-recording used for an ordinary
 * worker finish/fail report. Applies identically whether entry->queried is
 * true or false (Req 4.7: an unqueried finished extra reconciles through
 * "the same idempotent commit-then-apply path... used for queried jobs" --
 * there is no separate "assign" case for a finished outcome; record_job_
 * reconcile_assign_fact() is documented as running-only, see job.c's own
 * doc comment on it).
 *
 * finishedOk: bug-1747 fix (Req 4.2) -- routed through the NEW record_job_
 * finish_fact_at() (job.c), not on_job_finish()/record_job_finish_fact(),
 * so the job's recorded finish_time is the WORKER-reported entry->
 * finish_time_ms (Req 3.5), not reconciliation-processing wall-clock time.
 * record_job_finish_fact_at() expects the caller to already hold job's
 * write lock (like record_job_reconcile_pending_fact/assign_fact above),
 * unlike on_job_finish() which locks internally -- this function performs
 * that lookup/lock/unlock/dereference sequence itself, mirroring on_job_
 * finish()'s own internal sequence exactly (job.c). A jobid not found in
 * the job table is a silent no-op, mirroring on_job_finish()'s own
 * "can not find job" behavior for an unknown jobid.
 *
 * finishedFail: still routed through the existing, UNMODIFIED on_job_fail()
 * /record_job_fail_fact() -- Requirement 4.2 remains unsatisfiable for this
 * outcome via any existing path in this codebase, not a reconciliation-
 * specific gap: WAL_JOB_FAIL's payload (persist_submit_job_fact_ex(...,
 * WAL_JOB_FAIL, jobid, 0, WAL_JOB_FAIL_STAGE_FLAG|stage)) carries no
 * finish-time slot at any layer, and apply_job_fail_transition() (job.c)
 * unconditionally sets job->job.finish_time = time(NULL) at APPLY time
 * (not even receipt time) for EVERY fail-reporting caller in this system,
 * confirmed by direct inspection -- a pre-existing, systemic gap that
 * predates this feature and is out of this task's boundary to fix (see
 * buglog.json bug-1747, re-scoped). The reconciled job's STATE (EXIT) is
 * always correct either way; only its recorded finish_time reflects
 * apply-processing time rather than the worker's true report. */
static void resync_apply_finished_entry(cluster_t* c,
                                        xce_job_state_entry_t* entry)
{
    if (entry->status == JobStateStatus_finishedOk) {
        job_t* job = resync_lookup_job(c, entry->jobid);

        if (job != NULL) {
            xce_jobfinish_t finish
                = { .jobid = entry->jobid, .exit_code = 0 };

            lock(job, RTTI_LOCK_WRITE);
            record_job_finish_fact_at(c, job, &finish,
                                      entry->finish_time_ms);
            unlock(job);

            dereference(job);
        }
    } else {
        job_failinfo_t fail
            = { .jobid  = entry->jobid,
               .stage  = JOB_EXEC_STAGE_MAIN,
               .code   = entry->exit_code,
               .detail = (char*)"worker-rejoin-resync reconciliation" };
        on_job_fail(c, &fail);
    }
}

/* Req 4.1, 4.5, 4.6: routes an "unrecognized" entry through record_job_
 * reconcile_pending_fact() (task 2.1). Per Req 3.1/3.3, a worker only ever
 * reports "unrecognized" for a jobid it was explicitly asked about (an
 * unqueried extra is, by construction of Req 3.6's own "currently running
 * or recently finished" trigger, never reported unrecognized) -- so this
 * path is queried-only in practice; the caller (resync_apply_report below)
 * still clears the outstanding set uniformly for every queried entry
 * regardless of which branch handled it. */
static void resync_apply_unrecognized_entry(cluster_t* c,
                                            xce_job_state_entry_t* entry)
{
    job_t* job = resync_lookup_job(c, entry->jobid);
    int    rc;

    if (job == NULL)
        return;

    lock(job, RTTI_LOCK_WRITE);
    rc = record_job_reconcile_pending_fact(c, job);
    unlock(job);

    if (rc == (int)JOB_TRANSITION_APPLIED)
        persist_notify_job_updated_already_persisted(c, job);

    dereference(job);
}

int resync_apply_report(cluster_t* c, worker_t* w, uint64_t round_token,
                        xce_job_state_report_t* report)
{
    int i;

    if (c == NULL || w == NULL || report == NULL)
        return -EINVAL;
    if (report->count > 0 && report->entries == NULL)
        return -EINVAL;

    /* Req 6.1, 6.3: a report for a round already superseded/discarded is
     * dropped in its entirety, before anything else is touched -- checked
     * under w's own write lock (matching every other w->resync mutation
     * this function goes on to make, via resync_settle_entry, per entry
     * below). */
    lock(w, RTTI_LOCK_WRITE);
    if (!resync_round_is_current(w, round_token)) {
        unlock(w);
        return 0;
    }
    unlock(w);

    /* Per-entry routing never holds w's write lock while also locking a
     * job's own write lock -- mirrors dispatch_a_job_'s established
     * discipline (job.c: every branch releases w's lock BEFORE acquiring a
     * job's) rather than this file's other three functions' "caller holds
     * w's lock for the whole call" convention, which fits their simple,
     * single-object, single-call-site mutations but not this function's
     * fan-out into per-job job.c calls, each requiring its own job-level
     * lock. w's lock is only (re)acquired, narrowly, inside
     * resync_settle_entry's own outstanding-set/maybe-resolve mutation. */
    for (i = 0; i < report->count; i++) {
        xce_job_state_entry_t* entry = report->entries[i];

        if (entry == NULL)
            continue;

        switch (entry->status) {
        case JobStateStatus_running:
            resync_apply_running_entry(c, w, entry);
            break;
        case JobStateStatus_finishedOk:
        case JobStateStatus_finishedFail:
            resync_apply_finished_entry(c, entry);
            break;
        case JobStateStatus_unrecognized:
        default:
            resync_apply_unrecognized_entry(c, entry);
            break;
        }

        if (entry->queried)
            resync_settle_entry(c, w, round_token, entry->jobid);
    }

    return 0;
}

#ifdef XCE_TESTING
/* worker-rejoin-resync task 3.1 test seam (resync.h): builds a
 * resync_timeout_ctx_t exactly as resync_arm_round's own delay_task_ms call
 * would have (heap-allocated, since resync_on_timeout unconditionally
 * free()s it -- matching that exact contract here, not a stack copy, or
 * the free() below would be undefined behavior) and invokes resync_on_
 * timeout directly, so a test can treat "the timeout fired" as a plain,
 * synchronous function call -- no real delaymgr thread, no wall-clock
 * wait. */
void resync_test_fire_timeout(cluster_t* c, worker_t* w, uint64_t token)
{
    resync_timeout_ctx_t* ctx
        = (resync_timeout_ctx_t*)calloc(1, sizeof(*ctx));

    if (ctx == NULL)
        return;

    ctx->c           = c;
    ctx->w           = w;
    ctx->round_token = token;
    ctx->timeout_ms  = 0;

    resync_on_timeout(ctx);
}
#endif