myxce/src/scheduler/testsuite/test_sim_round_boundary_reconciliation.c

/*
 * test_sim_round_boundary_reconciliation.c - end-to-end integration proof
 * for scheduler-sim-decision-parity task 3 (Req 1.3, 1.4, 2.1, 2.2, 5.2):
 * Recorded-Decision Reconciliation (sim_reconcile_unaccounted_jobs(),
 * tasks 2.2-2.4) wired into sim_loop.c's round-boundary sequence, strictly
 * before sched_account_jobs() and xce_core_signal_go().
 *
 * Why this file exists, distinct from the tests that already cover the
 * pieces: test_sim_reconcile_unaccounted.c (tasks 2.2-2.4) already proves
 * sim_reconcile_unaccounted_jobs() itself is correct in isolation, calling
 * it directly against hand-built fixtures. test_sim_session_ordering.c
 * (task 2.1) already proves sched_account_jobs() is called once per round
 * inside the session-active window, via a spy. test_sim_resource_
 * contention.c (task 1.2) proves a resource-contention `.replay` fixture's
 * shape, with sched_account_jobs() stubbed to a no-op (it deliberately does
 * not link scheduler.c/resource.c). NONE of these exercises the actual
 * WIRED call site inside the real, unmodified sched_sim_driver_run() with
 * the REAL accounting/capacity-check code linked in -- which is exactly
 * what task 3's own Observable Completion requires: "a job whose
 * allocation was reconciled this round is reflected in resource-in-use
 * totals before the live picker/allocator evaluates capacity for any
 * other job in that same round." This file drives sched_sim_driver_run()
 * for real, against a hand-crafted `.replay` file (same technique
 * test_sim_verify.c's own header comment documents: sched_record_start()
 * for a real header+snapshot prefix, then hand-constructed ReplayEvent/
 * ReplaySession messages appended via the real encode_ReplayEvent_ptr()/
 * encode_ReplaySession_ptr()/write_capnp_message() codec), then inspects
 * the REAL post-replay xce_sched_core_t/xce_sched_worker_t state directly.
 *
 * Scenario: worker "worker-a" has 10.0 total units of one resource
 * ("slots"). Job A (9301) is pushed PEND, then flipped to RUN by a
 * JOB_STATE_CHANGE event in the SAME round's event batch -- the signature
 * of a job already resolved before this replay's own live picker ever
 * reaches it (design.md's Recorded-Decision Reconciliation trigger
 * condition: RUN-state, job->accounts == NULL). This round's recorded
 * ReplaySession carries a matching DISPATCH decision for job A (8.0
 * units) plus a genuine PEND decision for job B (9302, still PEND this
 * round -- the real picker independently re-pends it fresh, needing no
 * reconciliation). After replay: (a) worker-a's resInUse must show job
 * A's 8.0-unit charge (Req 1.3/1.4) -- proving reconciliation's real
 * make_job_account()/sched_charge_account() round trip, INCLUDING the
 * lazy resname->residx resolution sched_update_res_accounts() performs
 * (make_res_account(), xcealloc.c, always produces residx == -1); (b)
 * sched_check_host_satisfy() (resource.c, unmodified) against this SAME
 * post-round worker state must now correctly DENY a hypothetical job
 * wanting more than the 2.0-unit remaining headroom, and GRANT one wanting
 * exactly 2.0 -- proving Requirement 2's capacity-awareness is a genuine
 * property of the now-correct resInUse, not merely "resInUse is some
 * nonzero number."
 *
 * Because reconciliation's real make_job_account()/make_res_account()
 * path is exercised (unlike test_sim_resource_contention.c's own stand-in
 * allocator, which builds xce_job_account_t by hand with residx already
 * resolved, deliberately bypassing this), this fixture pre-registers
 * "slots" in the core's resmgr via the real makeResource() and reads back
 * its assigned ->idx, rather than assuming which index the lazy
 * add_vres() auto-registration path (resource.c) would land on -- see
 * fixture_init()'s own comment below.
 *
 * White-box #include set: the same combined core/sim/recording/repl/
 * snapshot stack test_sim_resource_contention.c's own header comment
 * documents, PLUS a REAL (not no-op) sched_account_jobs() reimplementation
 * below -- copied verbatim from scheduler.c:1258 (body unchanged, this
 * spec's own task 1.1 already relocated it to take xce_sched_core_t*
 * directly), following the SAME "real reimplementation of a scheduler.c/
 * schedobj.c function, to avoid pulling in the whole file's heavier
 * dependencies" convention this whole test-file family already uses for
 * sched_resolve_job_picker(). This is the one thing that MUST differ from
 * every other sibling file that #includes sim_loop.c: this file's whole
 * purpose is to prove what sched_account_jobs() actually does to
 * resInUse, so stubbing it as a no-op (as every other sim_loop.c-including
 * file in this directory does) would defeat the test.
 */
#include <errno.h>
#include <limits.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <unity/unity.h>

#include "controller.capnp.h"
#include "kvmgr.h"
#include "preason.h"
#include "resobj.h"
#include "scheduler.h"
#include "xstring.h"

/* Pull in the real functions under test into this TU (white-box #include,
 * same combined set test_sim_resource_contention.c's own header comment
 * documents -- driver.c/recording_events.c are deliberately NOT included
 * here: unlike that file, this one never drives a live normal-mode round
 * (no dataq/pend_updates machinery needed) -- it builds its `.replay` file
 * by hand, the same technique test_sim_verify.c's own header comment
 * documents, so it needs only recording.c/recording_session.c for the
 * header+snapshot prefix). */
#include "../src/core.c"
#include "../src/core_clock.c"
#include "../src/core_events.c"
#include "../src/core_session.c"
#include "../src/decision.c"
#include "../src/plugin.c"
#include "../src/plugins/builtin_fcfs.c"
#include "../src/recording.c"
#include "../src/recording_session.c"
#include "../src/repl.c"
#include "../src/repl_break.c"
#include "../src/repl_info.c"
#include "../src/repl_params.c"
#include "../src/resource.c"
#include "../src/sim.c"
#include "../src/sim_loop.c"
#include "../src/sim_picker.c" /* bug-1567: sched_sim_driver_run() calls sim_bootstrap_picker() (sim_loop.c-cap sibling) as an extern -- undefined reference at link time otherwise */
#include "../src/sim_record.c"
#include "../src/sim_verify.c"
#include "../src/snapshot.c"
#include "../src/snapshot_load.c"
#include "../src/snapshot_wire.c"

void kill_sched_job(void* arg) { (void)arg; }
void kill_sched_queue(void* arg) { (void)arg; }
void kill_sched_worker(void* arg) { (void)arg; }
void kill_sched_user(void* arg) { (void)arg; }
void kill_sched_project(void* arg) { (void)arg; }
void kill_sched_resval(xce_val_t* val) { (void)val; }

/* resource.c's sched_try_alloc_on_host_()'s own link-time dependency --
 * never actually invoked (this file calls sched_check_host_satisfy()
 * directly, never sched_try_alloc_on_hosts()/sched_try_alloc_on_host_()
 * itself). Same link-satisfying-only stub test_sim_resource_contention.c's
 * own resource.c inclusion already establishes needing.
 *
 * sched_worker_release_host_ctx/sched_fill_limit_requests are deliberately
 * NOT stubbed here (resource.c defines both for real, and this file's
 * combined include set already provides plugin_prepare_allocation via
 * plugin.c) -- same rationale test_sim_resource_contention.c's own header
 * comment documents. */
int sched_ctx_get_uncomit_host(xvec_t** v, xce_sched_ctx_t* c, int i)
{
    (void)v;
    (void)c;
    (void)i;
    return 0;
}

/* Real reimplementation of schedobj.c's sched_resolve_job_picker -- same
 * copy every white-box test in this directory that does not link
 * schedobj.c already establishes. */
int sched_resolve_job_picker(xce_sched_core_t* core, xce_sched_plugin_t** out)
{
    xce_sched_plugin_t* plugin = NULL;
    int                 rc     = 0;

    if ((core == NULL) || (out == NULL)) {
        return -EINVAL;
    }

    *out = NULL;

    if (core->job_pick_plugin == NULL) {
        return -EINVAL;
    }

    lock(core->plugins, RTTI_LOCK_READ);
    rc = key_find(core->plugins, core->job_pick_plugin, (void**)&plugin);
    unlock(core->plugins);

    if (rc <= 0) {
        return (rc < 0) ? rc : -ENOENT;
    }

    if ((plugin == NULL) || (plugin->picker_iface == NULL)
        || (plugin->picker_iface->next_job == NULL)
        || (plugin->picker_iface->begin_session == NULL)
        || (plugin->picker_iface->end_session == NULL)
        || (plugin->picker_iface->on_dispatch == NULL)
        || (plugin->picker_iface->on_pend == NULL)) {
        return -EINVAL;
    }

    *out = plugin;

    return 0;
}

/* Real reimplementation of scheduler.c's sched_account_jobs() (scheduler.c:
 * 1258, scheduler-sim-decision-parity task 1.1's relocated xce_sched_core_t*
 * signature) -- body copied verbatim, unlike every OTHER sibling test file
 * in this directory that #includes sim_loop.c (all of which stub this as a
 * no-op, since none of them need real accounting). This file's entire
 * purpose is to prove what this function does to resInUse once task 3
 * wires reconciliation ahead of it, so a no-op stub would defeat the
 * test's own premise. Calls the REAL sched_charge_account() (resource.c,
 * genuinely linked into this TU), not a stand-in. */
int sched_account_jobs(xce_sched_core_t* core)
{
    iterator_t it;
    int        rc = 0;

    rc = iter_make(&it, core->jobs);
    if (rc < 0) {
        return rc;
    }

    for (;;) {
        xce_sched_job_t* job = NULL;

        rc = iter_next(&it, (void**)&job);
        if ((rc < 0) || (rc == RTTI_ITER_NO_DATA)) {
            break;
        }

        if (!JOB_IS_RUN(job->state)) {
            continue;
        }

        if (job->accounts != NULL) {
            sched_charge_account(core, job->accounts);
        }
    }

    if (rc == RTTI_ITER_NO_DATA) {
        rc = 0;
    }

    iter_destroy(&it);

    return rc;
}

void setUp(void) { return; }
void tearDown(void) { return; }

/* ---------------------------------------------------------------------- */
/* Fixture constants                                                      */
/* ---------------------------------------------------------------------- */

#define WORKER_NAME     "worker-a"
#define WORKER_CAPACITY 10.0f

#define JOBA_ID   9301ULL /* already resolved -- reconciled this round */
#define JOBB_ID   9302ULL /* genuinely still PEND this round */
#define JOBA_WANT 8.0f

/* ---------------------------------------------------------------------- */
/* Fixture helpers                                                        */
/* ---------------------------------------------------------------------- */

static void make_tmp_dir(char* out, size_t out_len, const char* suffix)
{
    char tmpl[PATH_MAX];

    snprintf(tmpl, sizeof(tmpl), "/tmp/sched_simreconcile%s_XXXXXX", suffix);
    TEST_ASSERT_NOT_NULL(mkdtemp(tmpl));
    TEST_ASSERT_TRUE(strlen(tmpl) < out_len);
    strcpy(out, tmpl);
}

/* Builds a real xce_scheduler_t fixture to drive sched_record_start() (same
 * technique test_sim_verify.c's own build_fixture_scheduler() establishes --
 * no dataq/pend_updates needed since this file never drives a live
 * normal-mode round, only hand-appends ReplayEvent/ReplaySession messages)
 * PLUS a real worker "worker-a" with 10.0 finite units of one resource,
 * added directly via sched_add_worker() so the worker (with its resvals)
 * is captured into sched_record_start()'s own initial snapshot -- the SAME
 * "worker added directly, captured by the initial snapshot" technique
 * test_sim_resource_contention.c's own dual-use replay test already proves
 * works.
 *
 * Unlike test_sim_resource_contention.c's own fixture (which bypasses
 * resmgr name resolution entirely, setting residx directly on hand-built
 * xce_job_account_t structures), THIS file's reconciliation path goes
 * through the REAL make_job_account()/make_res_account() (xcealloc.c),
 * which always produces residx == -1 -- sched_update_res_accounts()
 * (resource.c) then resolves it lazily via key_find(core->resmgr,
 * resname, ...), auto-registering the resource via add_vres() if not
 * already present. Rather than assume which index that lazy path would
 * land on, this fixture pre-registers "slots" via the real makeResource()
 * and reads back its assigned ->idx, then uses THAT same index for the
 * worker's resvals/resInUse setup and for every capacity-check `wanted`
 * vector built later in the test -- eliminating any dependency on
 * add_vres()'s own index-assignment behavior.
 */
static int fixture_init(xce_scheduler_t* sched, char* record_dir)
{
    xce_sched_worker_t* w;
    xce_resource_t*     res_template;
    xce_resource_t*     registered;
    int                 rc;
    int                 residx;
    float               val[1];

    memset(sched, 0, sizeof(*sched));
    mylock_init(&(sched->record_lock));

    rc = xce_sched_core_init(&(sched->core));
    TEST_ASSERT_EQUAL_INT(0, rc);

    sched->core.job_pick_plugin = (char*)"builtin_fcfs";
    sched->record_dir           = record_dir;

    res_template = (xce_resource_t*)calloc(1, sizeof(*res_template));
    TEST_ASSERT_NOT_NULL(res_template);
    res_template->name    = strdup("slots");
    res_template->valtype = XCE_VAL_NUMERIC;
    res_template->ordtype = XCE_ORD_DECR;
    res_template->flags   = XCE_RESF_LIMIT | XCE_RESF_CONSUMABLE
                          | XCE_RESF_RELEASE | XCE_RESF_DEFAULT_HOST;

    registered = makeResource(sched->core.resmgr, res_template);
    TEST_ASSERT_NOT_NULL(registered);
    residx = registered->idx;

    w = (xce_sched_worker_t*)calloc(1, sizeof(*w));
    TEST_ASSERT_NOT_NULL(w);
    w->name = strdup(WORKER_NAME);
    TEST_ASSERT_NOT_NULL(w->name);

    rc = zarray_init(&(w->resvals), 8, NULL);
    TEST_ASSERT_EQUAL_INT(0, rc);
    rc = zarray_init(&(w->resInUse), 8, NULL);
    TEST_ASSERT_EQUAL_INT(0, rc);

    val[0] = WORKER_CAPACITY;
    rc     = put_resvals(&(w->resvals), 1, &residx, val);
    TEST_ASSERT_TRUE(rc >= 0);

    rc = sched_add_worker(&(sched->core), w);
    TEST_ASSERT_TRUE(rc >= 0);

    return residx;
}

/* Appends one ReplayEvent{queueUpdate} message -- same technique
 * test_sim_verify.c's own append_queue_update_event() establishes. */
static void append_queue_update_event(FILE* f, uint64_t ts,
                                      const char* queue_name)
{
    xce_queue_t          queue;
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr      p;

    memset(&queue, 0, sizeof(queue));
    queue.queue    = (char*)queue_name;
    queue.status   = QUEUE_STAT_OPEN;
    queue.priority = 1;

    memset(&src, 0, sizeof(src));
    src.ts                = ts;
    src.body_which        = ReplayEvent_body_queueUpdate;
    src.body.queue_update = &queue;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

/* Appends one ReplayEvent{jobsPush} message with exactly one JOB_STAT_PEND
 * job -- same technique test_sim_verify.c's own append_jobs_push_event()
 * establishes. */
static void append_jobs_push_event(FILE* f, uint64_t ts, uint64_t jobid,
                                   const char* queue_name)
{
    job_submit_data_t    submit;
    xce_job_t            job;
    xce_job_t*           jobs[1];
    xce_jobs_data_t      jobs_data;
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr      p;

    memset(&submit, 0, sizeof(submit));
    submit.queue    = (char*)queue_name;
    submit.user     = (char*)"alice";
    submit.mintasks = 1;
    submit.maxtasks = 1;

    memset(&job, 0, sizeof(job));
    job.jobid  = jobid;
    job.state  = JOB_STAT_PEND;
    job.submit = &submit;

    jobs[0] = &job;

    memset(&jobs_data, 0, sizeof(jobs_data));
    jobs_data.n_total = 1;
    jobs_data.count   = 1;
    jobs_data.jobs    = jobs;

    memset(&src, 0, sizeof(src));
    src.ts             = ts;
    src.body_which     = ReplayEvent_body_jobsPush;
    src.body.jobs_push = &jobs_data;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

/* Appends one ReplayEvent{jobStateChange} message -- the mechanism
 * (core_events.c's XCE_CORE_EVENT_JOB_STATE_CHANGE handling) that flips a
 * job to RUN state during replay independently of any decision, leaving
 * job->accounts/job->allocs untouched (still NULL) -- exactly the trigger
 * shape Recorded-Decision Reconciliation looks for. Same technique
 * test_sim_event_loop.c's own append_jobstatechange_event() establishes. */
static void append_job_state_change_event(FILE* f, uint64_t ts,
                                          uint64_t jobid, uint32_t old_state,
                                          uint32_t new_state)
{
    job_state_change_event_t jsc;
    replay_event_t           src;
    struct capn              c;
    struct capn_segment*     cs;
    ReplayEvent_ptr          p;

    memset(&jsc, 0, sizeof(jsc));
    jsc.jobid     = jobid;
    jsc.old_state = old_state;
    jsc.new_state = new_state;
    jsc.exit_code = 0;

    memset(&src, 0, sizeof(src));
    src.ts                    = ts;
    src.body_which            = ReplayEvent_body_jobStateChange;
    src.body.job_state_change = &jsc;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

/* Appends the .replay stream's whole-stream termination marker
 * (ReplayEvent{endSession}) -- same technique test_sim_verify.c's own
 * append_end_of_stream_marker() establishes. */
static void append_end_of_stream_marker(FILE* f)
{
    replay_event_t       src;
    struct capn          c;
    struct capn_segment* cs;
    ReplayEvent_ptr      p;

    memset(&src, 0, sizeof(src));
    src.body_which = ReplayEvent_body_endSession;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplayEvent_ptr(cs, &p, &src);
    capn_setp(capn_root(&c), 0, p.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

/* Builds the wire-shaped xce_job_alloc_t a DISPATCH decision carries --
 * same field shape test_sim_verify.c's own make_test_alloc()/
 * make_test_host_alloc()/make_test_job_alloc() establish. */
static xce_job_alloc_t* make_dispatch_alloc(uint64_t jobid, const char* host,
                                            float value)
{
    xce_alloc_t*       res    = (xce_alloc_t*)calloc(1, sizeof(*res));
    xce_host_alloc_t*  h      = (xce_host_alloc_t*)calloc(1, sizeof(*h));
    xce_host_alloc_t** hosts  = (xce_host_alloc_t**)calloc(1, sizeof(*hosts));
    xce_alloc_t**      allocs = (xce_alloc_t**)calloc(1, sizeof(*allocs));
    xce_job_alloc_t*   j      = (xce_job_alloc_t*)calloc(1, sizeof(*j));

    TEST_ASSERT_NOT_NULL(res);
    TEST_ASSERT_NOT_NULL(h);
    TEST_ASSERT_NOT_NULL(hosts);
    TEST_ASSERT_NOT_NULL(allocs);
    TEST_ASSERT_NOT_NULL(j);

    res->resname  = strdup("slots");
    res->reserve  = strdup("run");
    res->instid   = 1;
    res->value    = value;
    res->from     = 0;
    res->duration = 0;

    allocs[0] = res;

    h->host    = strdup(host);
    h->nallocs = 1;
    h->allocs  = allocs;

    hosts[0] = h;

    j->jobid         = jobid;
    j->nhosts        = 1;
    j->allocs        = hosts;
    j->nres_shared   = 0;
    j->allocs_shared = NULL;

    return j;
}

/* Appends one ReplaySession carrying a DISPATCH decision (job A) and a
 * PEND decision (job B) -- same encode_ReplaySession_ptr() technique
 * test_sim_verify.c's own append_replay_session_with_pend() establishes. */
static void append_replay_session_with_dispatch_and_pend(
    FILE* f, uint64_t begin_round, xce_job_alloc_t* dispatch_alloc,
    uint64_t pend_jobid, int pend_main_reason)
{
    sched_dispatch_decision_t dispatch;
    xce_job_preason_t        reason;
    sched_pend_decision_t    pend;
    replay_decision_t        d_dispatch;
    replay_decision_t        d_pend;
    replay_decision_t*       decisions[2];
    replay_session_t         session;
    struct capn              c;
    struct capn_segment*     cs;
    ReplaySession_ptr        sp;

    memset(&dispatch, 0, sizeof(dispatch));
    dispatch.alloc = dispatch_alloc;

    memset(&reason, 0, sizeof(reason));
    reason.main_reason = pend_main_reason;

    memset(&pend, 0, sizeof(pend));
    pend.jobid  = pend_jobid;
    pend.reason = &reason;

    memset(&d_dispatch, 0, sizeof(d_dispatch));
    d_dispatch.ts           = begin_round;
    d_dispatch.body_which   = ReplayDecision_body_dispatch;
    d_dispatch.body.dispatch = &dispatch;

    memset(&d_pend, 0, sizeof(d_pend));
    d_pend.ts         = begin_round;
    d_pend.body_which = ReplayDecision_body_pend;
    d_pend.body.pend  = &pend;

    decisions[0] = &d_dispatch;
    decisions[1] = &d_pend;

    memset(&session, 0, sizeof(session));
    session.begin_round = begin_round;
    session.decisions   = decisions;
    session.n_decisions = 2;
    session.summary     = NULL;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_ReplaySession_ptr(cs, &sp, &session);
    capn_setp(capn_root(&c), 0, sp.p);

    TEST_ASSERT_EQUAL_INT(0, write_capnp_message(f, &c));
    capn_free(&c);
}

/* ---------------------------------------------------------------------- */
/* The end-to-end proof (task 3's own Observable Completion).             */
/* ---------------------------------------------------------------------- */

/*
 * test_reconciled_job_charges_resinuse_before_live_capacity_check - builds
 * a `.replay` fixture where job A is already RUN with no live decision this
 * round (reconciliation's trigger shape) and job B is genuinely still
 * PEND, replays it through the REAL, unmodified sched_sim_driver_run(),
 * then proves:
 *
 *   (a) worker-a's resInUse reflects job A's 8.0-unit charge immediately
 *       after this SAME round (Req 1.3, 1.4) -- reconciliation ran, built
 *       job->accounts via the real make_job_account(), and
 *       sched_account_jobs() (called strictly after reconciliation, per
 *       task 3's own ordering) charged it into resInUse before this
 *       function returns.
 *
 *   (b) sched_check_host_satisfy() (resource.c, completely unmodified)
 *       against this SAME post-round worker state now correctly DENIES a
 *       hypothetical job wanting more than the 2.0-unit remaining
 *       headroom, and GRANTS one wanting exactly 2.0 -- proving
 *       Requirement 2's capacity-awareness is a genuine, working property
 *       of the now-correct resInUse (not merely that resInUse holds some
 *       nonzero number).
 *
 * Before task 3's wiring, sim_reconcile_unaccounted_jobs() was never
 * called, so job A's RUN state was never reconciled: job->accounts stays
 * NULL, sched_account_jobs() skips it (its own job->accounts != NULL
 * guard), and resInUse for worker-a stays at 0.0 for the whole run --
 * this is bug-1613/1611's own documented behavior, and exactly what
 * this test's assertion (a) would catch as a regression if the wiring
 * were ever removed.
 */
static void
test_reconciled_job_charges_resinuse_before_live_capacity_check(void)
{
    char                   record_dir[PATH_MAX];
    xce_scheduler_t        sched;
    xce_scheduler_t        wrapper;
    FILE*                  f;
    char                   r_path[PATH_MAX];
    int                    residx;
    xce_job_alloc_t*       alloc;
    sched_sim_args_t       sim_args;
    xce_sched_sim_driver_t sim;
    xce_sched_worker_t*    w;
    xce_sched_job_t*       jobA;
    xce_sched_job_t*       jobB;
    int                    rc;
    float                  got[1];
    xvec_t                 uncomitted;
    xvec_idval_t           wanted;
    xvec_idval_t           unmeet;

    make_tmp_dir(record_dir, sizeof(record_dir), "_e2e");
    residx = fixture_init(&sched, record_dir);

    rc = sched_record_start(&sched, "reconciliation.replay");
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_TRUE(strlen(sched.record_file_path) < sizeof(r_path));
    strcpy(r_path, sched.record_file_path);
    f = sched.record_file;

    append_queue_update_event(f, 1000ULL, "normal");
    append_jobs_push_event(f, 1500ULL, JOBA_ID, "normal");
    append_jobs_push_event(f, 1500ULL, JOBB_ID, "normal");
    append_job_state_change_event(f, 1800ULL, JOBA_ID, JOB_STAT_PEND,
                                  JOB_STAT_RUN);

    alloc = make_dispatch_alloc(JOBA_ID, WORKER_NAME, JOBA_WANT);
    append_replay_session_with_dispatch_and_pend(f, 2000ULL, alloc, JOBB_ID,
                                                 0);
    append_end_of_stream_marker(f);
    fclose(f);

    memset(&sim_args, 0, sizeof(sim_args));
    sim_args.replay_path  = r_path;
    sim_args.sim_start_ts = SCHED_SIM_START_TS_LATEST;
    sim_args.sim_speed    = 0; /* frozen -- deterministic, no wall-clock skew */

    memset(&sim, 0, sizeof(sim));
    rc = sched_sim_driver_run(&sim, &sim_args);
    TEST_ASSERT_EQUAL_INT(0, rc);

    /* Sanity: job A really is RUN with a reconstructed allocation, job B
     * really is still PEND with none -- the fixture's own shape proof,
     * not merely trusting the input constants. */
    jobA = sched_get_job(&(sim.core), JOBA_ID);
    TEST_ASSERT_NOT_NULL(jobA);
    TEST_ASSERT_TRUE(JOB_IS_RUN(jobA->state));
    TEST_ASSERT_NOT_NULL_MESSAGE(
        jobA->accounts,
        "reconciliation must have reconstructed job A's accounts from the "
        "recorded DISPATCH decision");

    jobB = sched_get_job(&(sim.core), JOBB_ID);
    TEST_ASSERT_NOT_NULL(jobB);
    TEST_ASSERT_FALSE(JOB_IS_RUN(jobB->state));
    TEST_ASSERT_NULL(jobB->accounts);

    /* (a) Req 1.3/1.4: worker-a's resInUse reflects job A's charge this
     * SAME round. */
    w = sched_get_worker(&(sim.core), WORKER_NAME);
    TEST_ASSERT_NOT_NULL(w);

    got[0] = -1.0f;
    rc     = get_resvals(&(w->resInUse), 1, &residx, got);
    TEST_ASSERT_EQUAL_INT(1, rc); /* get_resvals() returns nres on success */
    TEST_ASSERT_EQUAL_FLOAT_MESSAGE(
        JOBA_WANT, got[0],
        "worker-a's resInUse must reflect job A's reconciled 8.0-unit "
        "charge before this function returns");

    /* (b) Req 2.1/2.2: the REAL, unmodified sched_check_host_satisfy()
     * against this SAME post-round state must now correctly deny a job
     * wanting more than the 2.0-unit remaining headroom (10.0 total -
     * 8.0 charged), and grant one wanting exactly the remainder. sched
     * is only used via sched->core.resmgr inside resource.c's own
     * unsatisfy_() TRACE line (design.md's own confirmed-by-grep note),
     * so a throwaway xce_scheduler_t wrapping the SAME sim.core is a
     * valid, safe caller-owned argument here -- not a special case. */
    memset(&wrapper, 0, sizeof(wrapper));
    wrapper.core = sim.core;

    rc = xvec_init(&uncomitted, 1, sizeof(float));
    TEST_ASSERT_TRUE(rc >= 0);
    rc = xvec_idval_init(&wanted, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);
    rc = xvec_idval_store(&wanted, residx, 5.0f); /* > 2.0 remaining */
    TEST_ASSERT_TRUE(rc >= 0);

    rc = sched_check_host_satisfy(&wrapper, w, &uncomitted, &wanted, &unmeet);
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, rc,
        "a job wanting more than the 2.0-unit remaining headroom must be "
        "denied -- capacity awareness must hold against the reconciled "
        "resInUse, not just report a number");

    xvec_fini(&uncomitted);
    xvec_idval_fini(&wanted);
    xvec_idval_fini(&unmeet);

    rc = xvec_init(&uncomitted, 1, sizeof(float));
    TEST_ASSERT_TRUE(rc >= 0);
    rc = xvec_idval_init(&wanted, 1);
    TEST_ASSERT_EQUAL_INT(0, rc);
    rc = xvec_idval_store(&wanted, residx, 2.0f); /* == 2.0 remaining, exact fit */
    TEST_ASSERT_TRUE(rc >= 0);

    rc = sched_check_host_satisfy(&wrapper, w, &uncomitted, &wanted, &unmeet);
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        1, rc,
        "a job wanting exactly the 2.0-unit remaining headroom must be "
        "granted -- proving the denial above is genuine capacity "
        "arithmetic, not a fixture that always denies");

    xvec_fini(&uncomitted);
    xvec_idval_fini(&wanted);
    xvec_idval_fini(&unmeet);

    printf("test_reconciled_job_charges_resinuse_before_live_capacity_"
          "check: PASS\n");
}

int main(void)
{
    UNITY_BEGIN();
    RUN_TEST(test_reconciled_job_charges_resinuse_before_live_capacity_check);
    return UNITY_END();
}