myxce/src/controller/testsuite/test_wal_job_dispatch_fact.c

/* test_wal_job_dispatch_fact.c — scheduler-dispatch-commit task 1.2
 *
 * Proves the new WAL_JOB_DISPATCH fact type (wal_types.h) -- the durable
 * record of a job's dispatch-time transition to a running state (Req 2.4,
 * design.md "WAL Dispatch Fact") -- is wired into BOTH of persist.c's apply
 * seams:
 *
 *   - apply_committed_entry() (live-commit apply / boot replay dispatch)
 *   - apply_wal_entry()       (standby catch-up / replicated-apply dispatch)
 *
 * and that appending ONLY a WAL_JOB_DISPATCH fact (no accompanying
 * WAL_JOB_START) and replaying it -- simulating a controller restart via a
 * real persist_recover() over a real on-disk WAL -- recovers the job as
 * running (this task's "Observable completion").
 *
 * This task does NOT wire any live caller that appends this fact type
 * (dispatch_a_job_() is untouched -- that's task 3.1's job); every test here
 * constructs and applies/replays a WAL_JOB_DISPATCH entry directly.
 *
 * Fixtures mirror this suite's established conventions exactly:
 *   - wjd_setup/wjd_teardown: bare cluster_t (jobmgr + hooksys) + in-memory
 *     persist_t, mirroring test_apply_committed_entry.c's tace_setup and
 *     test_job_array_apply.c's tajd_setup.
 *   - wjd_make_fresh_persist/wjd_teardown_persist: real disk-backed
 *     persist_t + persist_recover(), mirroring test_replay_apply_seam.c's
 *     trr_make_fresh_persist/trr_teardown_persist.
 */
#include <rtti.h>

#include <capnp_c.h>
#include <errno.h>
#include <stdarg.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <unity/unity.h>

#include "cluster.h"
#include "controller.capnp.h"
#include "hook.h"
#include "jobj.h"
#include "persist.h"
#include "wal_types.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xcealloc.h"
#include "xstring.h"

/* jobj.c's jobsub table and the process-global xstring table are shared
 * across every test file in the controller_unit_test binary — guard
 * idempotently, mirroring every sibling fixture in this suite. */
static int g_wjd_jobobj_ready;

static cluster_t g_wjd_c;
static persist_t g_wjd_p;

/* ── Spy hook: counts AFTER_EVENT firings, ignores all payload args ─────── */
static int g_wjd_hook_fires;

static int wjd_spy_hook(int type, int situation, int detail, void* data,
                        va_list ap)
{
    (void)type; (void)detail; (void)data; (void)ap;
    if (situation == HOOK_AFTER_EVENT)
        g_wjd_hook_fires++;
    return 0;
}

static void wjd_setup(void)
{
    kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
                                  .nslots   = 32,
                                  .delta    = 16 };

    if (makeString((char*)"wjd-probe") == NULL)
        xstring_table_init(64);

    if (!g_wjd_jobobj_ready) {
        TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
        g_wjd_jobobj_ready = 1;
    }

    memset(&g_wjd_c, 0, sizeof(g_wjd_c));
    TEST_ASSERT_EQUAL_INT(0,
        xce_hook_init(&g_wjd_c.hooksys, XCE_CONTROLLER_HOOK_MAX));

    g_wjd_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
    TEST_ASSERT_NOT_NULL(g_wjd_c.jobmgr);

    memset(&g_wjd_p, 0, sizeof(g_wjd_p));
    g_wjd_p.cluster = &g_wjd_c;

    g_wjd_hook_fires = 0;
}

static void wjd_teardown(void)
{
    xce_hook_fini(&g_wjd_c.hooksys);
    if (g_wjd_c.jobmgr)
        dereference(g_wjd_c.jobmgr);
}

/* Look up a job_t wrapper by jobid (NULL if absent). */
static job_t* wjd_find_job(uint64_t jobid)
{
    job_t* j = NULL;
    lock(g_wjd_c.jobmgr, RTTI_LOCK_READ);
    int found = key_find(g_wjd_c.jobmgr, (void*)jobid, (void**)&j);
    unlock(g_wjd_c.jobmgr);
    return (found > 0) ? j : NULL;
}

/* Creates a job directly in the fixture's jobmgr, bypassing dispatch/replay
 * entirely — for seeding fact-transition tests, mirroring
 * test_apply_committed_entry.c's tace_seed_job. */
static job_submit_data_t g_wjd_seed_submit;

static job_t* wjd_seed_job(uint64_t jobid, uint32_t state)
{
    xce_job_t jdef;

    memset(&g_wjd_seed_submit, 0, sizeof(g_wjd_seed_submit));
    g_wjd_seed_submit.queue = (char*)"q";
    g_wjd_seed_submit.user  = (char*)"tester";

    memset(&jdef, 0, sizeof(jdef));
    jdef.jobid  = jobid;
    jdef.state  = state;
    jdef.submit = &g_wjd_seed_submit;

    return makeJob(g_wjd_c.jobmgr, &jdef);
}

/* ── apply_committed_entry(): live-commit apply / boot replay dispatch ──── */

static void test_wjd_apply_committed_entry_transitions_pend_to_run(void)
{
    wjd_setup();

    job_t* j = wjd_seed_job(71001, JOB_STAT_PEND);
    TEST_ASSERT_NOT_NULL(j);

    wal_jobstate_t js = { .jobid = 71001, .state = JOB_STAT_RUN, ._pad = 0 };

    xce_add_hook(&g_wjd_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjd_spy_hook,
                NULL);

    int rc = apply_committed_entry(&g_wjd_p, WAL_JOB_DISPATCH, 100, &js,
                                   sizeof(js));
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state,
        "WAL_JOB_DISPATCH must apply via the same absolute-value guarded "
        "core as WAL_JOB_START/etc, joining their fall-through group");
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_wjd_hook_fires,
        "a forward WAL_JOB_DISPATCH transition must fire exactly one "
        "notification, matching every other type in its shared group");

    wjd_teardown();
    printf("test_wjd_apply_committed_entry_transitions_pend_to_run: PASS\n");
}

/* bug-1678: build a wal_jobstate_t header + trailing Cap'n Proto JobAlloc
 * bytes -- persist_submit_job_dispatch_fact()'s payload shape -- mirroring
 * capnp_encode_joballoc (persist.c, file-static) since job.c/persist.c's own
 * static helpers aren't linkable from a test TU. Caller owns *out. */
static int wjd_build_dispatch_payload(uint64_t jobid, uint32_t state,
                                      xce_job_alloc_t* alloc,
                                      uint8_t** out, size_t* out_len)
{
    struct capn          c;
    struct capn_segment* cs;
    struct JobAlloc      cja;
    JobAlloc_ptr         jap;
    uint8_t*             allocbuf;
    size_t               alloclen;
    wal_jobstate_t        js = { .jobid = jobid, .state = state, ._pad = 0 };

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_JobAlloc(cs, &cja, alloc);
    jap = new_JobAlloc(cs);
    write_JobAlloc(&cja, jap);
    capn_setp(capn_root(&c), 0, jap.p);

    alloclen = capn_size(&c);
    allocbuf = malloc(alloclen);
    if (!allocbuf) { capn_free(&c); return -ENOMEM; }
    capn_write_mem(&c, allocbuf, alloclen, 0);
    capn_free(&c);

    *out_len = sizeof(js) + alloclen;
    *out     = malloc(*out_len);
    if (!*out) { free(allocbuf); return -ENOMEM; }
    memcpy(*out, &js, sizeof(js));
    memcpy(*out + sizeof(js), allocbuf, alloclen);
    free(allocbuf);
    return 0;
}

/* One host ("wjd-worker"), one resource ("slots", value 1.0) -- the same
 * minimal shape test_dispatch_commit.c's own dispatch tests use. Caller owns
 * the result (release via free_JobAlloc_ptr). */
static xce_job_alloc_t* wjd_make_test_alloc(uint64_t jobid)
{
    return make_job_alloc(jobid, 1,
        make_host_allocs(1,
            make_host_alloc((char*)"wjd-worker", 1,
                make_allocs(1,
                    make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
}

static void test_wjd_apply_committed_entry_transitions_and_sets_allocs(void)
{
    wjd_setup();

    job_t* j = wjd_seed_job(71004, JOB_STAT_PEND);
    TEST_ASSERT_NOT_NULL(j);

    xce_job_alloc_t* alloc = wjd_make_test_alloc(71004);
    TEST_ASSERT_NOT_NULL(alloc);

    uint8_t* payload = NULL;
    size_t   plen    = 0;
    TEST_ASSERT_EQUAL_INT(0,
        wjd_build_dispatch_payload(71004, JOB_STAT_RUN, alloc, &payload,
                                   &plen));
    free_JobAlloc_ptr(&alloc);

    int rc = apply_committed_entry(&g_wjd_p, WAL_JOB_DISPATCH, 102, payload,
                                   (uint32_t)plen);
    free(payload);

    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state);
    TEST_ASSERT_NOT_NULL_MESSAGE(j->job.allocs,
        "a WAL_JOB_DISPATCH payload with trailing JobAlloc bytes must set "
        "job.allocs (bug-1678), not just state");
    TEST_ASSERT_EQUAL_INT(1, j->job.allocs->nhosts);
    TEST_ASSERT_EQUAL_STRING("wjd-worker", j->job.allocs->allocs[0]->host);
    TEST_ASSERT_EQUAL_STRING("slots",
        j->job.allocs->allocs[0]->allocs[0]->resname);

    wjd_teardown();
    printf("test_wjd_apply_committed_entry_transitions_and_sets_allocs: "
          "PASS\n");
}

/* ── apply_wal_entry(): standby catch-up / replicated-apply dispatch ────── */

static void
test_wjd_apply_wal_entry_transitions_pend_to_run_no_hook(void)
{
    wjd_setup();

    job_t* j = wjd_seed_job(71002, JOB_STAT_PEND);
    TEST_ASSERT_NOT_NULL(j);

    wal_jobstate_t js = { .jobid = 71002, .state = JOB_STAT_RUN, ._pad = 0 };

    xce_add_hook(&g_wjd_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjd_spy_hook,
                NULL);

    int rc = apply_wal_entry(&g_wjd_p, WAL_JOB_DISPATCH, 101, &js,
                             sizeof(js));
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state,
        "WAL_JOB_DISPATCH must apply via restore_job_state() through "
        "apply_wal_entry's catch-up dispatch, joining the same "
        "absolute-value fall-through group as WAL_JOB_START/etc");
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wjd_hook_fires,
        "apply_wal_entry's catch-up dispatch must never fire a notification "
        "hook, matching every other type in its group");

    wjd_teardown();
    printf("test_wjd_apply_wal_entry_transitions_pend_to_run_no_hook: "
          "PASS\n");
}

static void test_wjd_apply_wal_entry_transitions_and_sets_allocs(void)
{
    wjd_setup();

    job_t* j = wjd_seed_job(71005, JOB_STAT_PEND);
    TEST_ASSERT_NOT_NULL(j);

    xce_job_alloc_t* alloc = wjd_make_test_alloc(71005);
    TEST_ASSERT_NOT_NULL(alloc);

    uint8_t* payload = NULL;
    size_t   plen    = 0;
    TEST_ASSERT_EQUAL_INT(0,
        wjd_build_dispatch_payload(71005, JOB_STAT_RUN, alloc, &payload,
                                   &plen));
    free_JobAlloc_ptr(&alloc);

    int rc = apply_wal_entry(&g_wjd_p, WAL_JOB_DISPATCH, 103, payload,
                             (uint32_t)plen);
    free(payload);

    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state);
    TEST_ASSERT_NOT_NULL_MESSAGE(j->job.allocs,
        "apply_wal_entry's catch-up dispatch must also recover job.allocs "
        "from a WAL_JOB_DISPATCH payload's trailing JobAlloc bytes "
        "(bug-1678) -- required for a promoted standby, not just the "
        "primary's own boot replay");
    TEST_ASSERT_EQUAL_INT(1, j->job.allocs->nhosts);
    TEST_ASSERT_EQUAL_STRING("wjd-worker", j->job.allocs->allocs[0]->host);

    wjd_teardown();
    printf("test_wjd_apply_wal_entry_transitions_and_sets_allocs: PASS\n");
}

/* ── Restart-replay Observable (this task's completion criterion) ───────── */

static persist_t* wjd_make_fresh_persist(const char* tmpdir)
{
    persist_t* p = calloc(1, sizeof(persist_t));
    if (!p) return NULL;

    persist_config_t cfg = persist_config_defaults();
    cfg.wal_sync_ms = 5000; /* slow fsync — no races in tests */

    persist_set_workdir(p, tmpdir, "ctrl-test");

    if (persist_init(p, &cfg, NULL) != 0) {
        free(p);
        return NULL;
    }
    if (persist_start(p) != 0) {
        persist_fini(p);
        free(p);
        return NULL;
    }
    return p;
}

static void wjd_teardown_persist(persist_t* p)
{
    persist_stop(p);
    persist_fini(p);
    free(p);
}

/* Encode helper mirroring test_apply_committed_entry.c's tace_encode_job /
 * test_replay_apply_seam.c's trr_encode_job (capnp_encode_job is file-static
 * in persist.c). */
static int wjd_encode_job(const xce_job_t* j, uint8_t** out, size_t* out_len)
{
    struct capn          c;
    struct capn_segment* cs;
    struct Job           cj;
    Job_ptr              jp;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_Job(cs, &cj, (xce_job_t*)j);
    jp = new_Job(cs);
    write_Job(&cj, jp);
    capn_setp(capn_root(&c), 0, jp.p);

    *out_len = capn_size(&c);
    *out     = malloc(*out_len);
    if (!*out) { capn_free(&c); return -ENOMEM; }
    capn_write_mem(&c, *out, *out_len, 0);
    capn_free(&c);
    return 0;
}

/* Appends ONLY WAL_JOB_SUBMIT (jobid PEND) + WAL_JOB_DISPATCH -- no
 * WAL_JOB_START anywhere in this log -- then replays it via a fresh
 * persist_t's persist_recover() (a real restart). This is the task's own
 * "Observable completion": the dispatch-time transition survives a restart
 * standing entirely on its own, with no dependence on the separate
 * worker-confirmed-start fact. */
static void
test_wjd_restart_replay_recovers_running_without_start_fact(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_wjd_replay_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    /* Phase 1: write the log to real disk. No cluster needed —
     * persist_wal_async is a physical-write-only path. */
    {
        persist_t* p = wjd_make_fresh_persist(tmp);
        TEST_ASSERT_NOT_NULL(p);

        job_submit_data_t submit;
        memset(&submit, 0, sizeof(submit));
        submit.queue = (char*)"q";
        submit.user  = (char*)"tester";

        xce_job_t jdef;
        memset(&jdef, 0, sizeof(jdef));
        jdef.jobid  = 71003;
        jdef.state  = JOB_STAT_PEND;
        jdef.submit = &submit;

        uint8_t* job_payload = NULL;
        size_t   job_plen    = 0;
        TEST_ASSERT_EQUAL_INT(0,
            wjd_encode_job(&jdef, &job_payload, &job_plen));
        TEST_ASSERT_EQUAL_INT(0,
            persist_wal_async(p, WAL_JOB_SUBMIT, job_payload,
                              (uint32_t)job_plen));
        free(job_payload);

        /* The dispatch-time fact — no WAL_JOB_START anywhere in this log. */
        wal_jobstate_t js = { .jobid = 71003, .state = JOB_STAT_RUN,
                              ._pad = 0 };
        TEST_ASSERT_EQUAL_INT(0,
            persist_wal_async(p, WAL_JOB_DISPATCH, &js, sizeof(js)));

        TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
        wjd_teardown_persist(p);
    }

    /* Phase 2: fresh persist_t + fresh cluster_t, recover from disk —
     * simulates a controller restart. */
    {
        persist_t* p2 = wjd_make_fresh_persist(tmp);
        TEST_ASSERT_NOT_NULL(p2);

        wjd_setup();
        p2->cluster = &g_wjd_c;

        int rc = persist_recover(p2);
        TEST_ASSERT_EQUAL_INT(0, rc);

        job_t* j = wjd_find_job(71003);
        TEST_ASSERT_NOT_NULL_MESSAGE(j,
            "restart replay must construct the job via WAL_JOB_SUBMIT");
        TEST_ASSERT_TRUE_MESSAGE(JOB_IS_RUN(j->job.state),
            "restart replay must recover the job as RUNNING from the "
            "WAL_JOB_DISPATCH fact alone, with no WAL_JOB_START present "
            "in the log at all");
        TEST_ASSERT_TRUE_MESSAGE((j->job.state & JOB_STAT_UNKWN) != 0,
            "bug-1509 B3: a RUN job recovered on restart is marked UNKWN "
            "(in-memory) awaiting the worker reconnect/resend to reconcile "
            "the true state");

        wjd_teardown();
        wjd_teardown_persist(p2);
    }

    char cmd[512];
    snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp);
    if (system(cmd) != 0)
        fprintf(stderr, "cleanup warning: rm -rf %s failed\n", tmp);

    printf("test_wjd_restart_replay_recovers_running_without_start_fact: "
          "PASS\n");
}

/* bug-1678: sibling of the test above, proving the user's exact acceptance
 * criterion ("after recover the job should get the same as it's state
 * before shutdown") at this file's own direct-payload granularity: a
 * WAL_JOB_DISPATCH fact carrying trailing JobAlloc bytes survives a real
 * restart (persist_recover over real on-disk WAL) with job.allocs intact,
 * not just job.state. test_dispatch_commit.c's
 * test_dc_success_appends_wal_job_dispatch_durably covers the same
 * observable through the REAL dispatch_a_job_() production call site; this
 * test isolates it at the WAL fact level, mirroring how this file's other
 * tests already isolate apply_committed_entry/apply_wal_entry from the live
 * caller. */
static void
test_wjd_restart_replay_recovers_running_and_allocs_with_dispatch_fact(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_wjd_replay2_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    /* Phase 1: write the log to real disk. */
    {
        persist_t* p = wjd_make_fresh_persist(tmp);
        TEST_ASSERT_NOT_NULL(p);

        job_submit_data_t submit;
        memset(&submit, 0, sizeof(submit));
        submit.queue = (char*)"q";
        submit.user  = (char*)"tester";

        xce_job_t jdef;
        memset(&jdef, 0, sizeof(jdef));
        jdef.jobid  = 71006;
        jdef.state  = JOB_STAT_PEND;
        jdef.submit = &submit;

        uint8_t* job_payload = NULL;
        size_t   job_plen    = 0;
        TEST_ASSERT_EQUAL_INT(0,
            wjd_encode_job(&jdef, &job_payload, &job_plen));
        TEST_ASSERT_EQUAL_INT(0,
            persist_wal_async(p, WAL_JOB_SUBMIT, job_payload,
                              (uint32_t)job_plen));
        free(job_payload);

        xce_job_alloc_t* alloc = wjd_make_test_alloc(71006);
        TEST_ASSERT_NOT_NULL(alloc);

        uint8_t* dispatch_payload = NULL;
        size_t   dispatch_plen    = 0;
        TEST_ASSERT_EQUAL_INT(0,
            wjd_build_dispatch_payload(71006, JOB_STAT_RUN, alloc,
                                       &dispatch_payload, &dispatch_plen));
        free_JobAlloc_ptr(&alloc);

        TEST_ASSERT_EQUAL_INT(0,
            persist_wal_async(p, WAL_JOB_DISPATCH, dispatch_payload,
                              (uint32_t)dispatch_plen));
        free(dispatch_payload);

        TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
        wjd_teardown_persist(p);
    }

    /* Phase 2: fresh persist_t + fresh cluster_t, recover from disk —
     * simulates a controller restart. */
    {
        persist_t* p2 = wjd_make_fresh_persist(tmp);
        TEST_ASSERT_NOT_NULL(p2);

        wjd_setup();
        p2->cluster = &g_wjd_c;

        int rc = persist_recover(p2);
        TEST_ASSERT_EQUAL_INT(0, rc);

        job_t* j = wjd_find_job(71006);
        TEST_ASSERT_NOT_NULL_MESSAGE(j,
            "restart replay must construct the job via WAL_JOB_SUBMIT");
        TEST_ASSERT_TRUE_MESSAGE(JOB_IS_RUN(j->job.state),
            "restart replay must recover the job as RUNNING from the "
            "WAL_JOB_DISPATCH fact's state");
        TEST_ASSERT_TRUE_MESSAGE((j->job.state & JOB_STAT_UNKWN) != 0,
            "bug-1509 B3: a RUN job recovered on restart is marked UNKWN "
            "(in-memory) awaiting worker reconnect/resend reconciliation");
        TEST_ASSERT_NOT_NULL_MESSAGE(j->job.allocs,
            "restart replay must recover job.allocs from the "
            "WAL_JOB_DISPATCH fact's trailing JobAlloc bytes (bug-1678), "
            "not just job.state");
        TEST_ASSERT_EQUAL_INT(1, j->job.allocs->nhosts);
        TEST_ASSERT_EQUAL_STRING("wjd-worker",
            j->job.allocs->allocs[0]->host);

        wjd_teardown();
        wjd_teardown_persist(p2);
    }

    char cmd[512];
    snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp);
    if (system(cmd) != 0)
        fprintf(stderr, "cleanup warning: rm -rf %s failed\n", tmp);

    printf("test_wjd_restart_replay_recovers_running_and_allocs_with_"
          "dispatch_fact: PASS\n");
}

/* ── Test group entry point ───────────────────────────────────────────── */

void test_wal_job_dispatch_fact(void)
{
    UnitySetTestFile(__FILE__);
    RUN_TEST(test_wjd_apply_committed_entry_transitions_pend_to_run);
    RUN_TEST(test_wjd_apply_committed_entry_transitions_and_sets_allocs);
    RUN_TEST(test_wjd_apply_wal_entry_transitions_pend_to_run_no_hook);
    RUN_TEST(test_wjd_apply_wal_entry_transitions_and_sets_allocs);
    RUN_TEST(test_wjd_restart_replay_recovers_running_without_start_fact);
    RUN_TEST(
        test_wjd_restart_replay_recovers_running_and_allocs_with_dispatch_fact);
}