myxce/src/controller/testsuite/test_apply_committed_entry.c

/* test_apply_committed_entry.c — raft-apply-decoupling task 2.2
 *
 * Proves apply_committed_entry() (persist.c) is the single dispatch seam
 * that turns ONE committed WAL entry into an in-memory mutation for every
 * classified entry type, and fires that entry type's post-mutation
 * notification hook exactly once, strictly after the mutation completes:
 *
 *   - WAL_JOB_SUBMIT      -> restore_job() + HOOK_DETAIL_JOB_NEW
 *   - WAL_JOB_START/...   -> apply_job_update_transition() (2.1's guarded
 *                            core) + HOOK_DETAIL_JOB_UPDATE
 *   - WAL_QUEUE_CTRL      -> restore_queue_state() + HOOK_DETAIL_QUEUE_UPDATE
 *   - WAL_WORKER_CTRL     -> worker_from_capnp() + HOOK_DETAIL_WORKER_JOIN
 *
 * and that a guard-rejected or idempotent-duplicate application performs no
 * mutation and fires NO notification (Req 7.1/7.2).
 *
 * This task does NOT wire apply_committed_entry into any live call site
 * (replay / apply pump are later tasks 2.3/6.1) — every test here calls the
 * dispatch function directly.
 *
 * Fixture: a minimal cluster_t (hooksys + jobmgr/qmgr/workermgr) and a bare
 * persist_t whose ONLY field apply_committed_entry reads is ->cluster — no
 * real WAL/disk/repl/commit-engine setup needed (mirrors test_apply_seam.c's
 * "bare fixture, no more than the function under test needs" philosophy).
 */

#include <rtti.h>

#include <capnp_c.h>
#include <errno.h>
#include <stdarg.h>
#include <stdbool.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 "qobj.h"
#include "wal_types.h"
#include "wobj.h"
#include "xce/cluster_def.h"
#include "xce/types.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 test_apply_seam.c. */
static int g_tace_jobobj_ready;

static cluster_t g_tace_c;
static persist_t g_tace_p;

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

static int tace_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_tace_hook_fires++;
    return 0;
}

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

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

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

    memset(&g_tace_c, 0, sizeof(g_tace_c));
    g_tace_c.myhost           = &g_tace_c.myhost_info;
    g_tace_c.myhost_info.name = (char*)"ctrl-test";
    g_tace_c.myhost_info.key  = (char*)"";

    TEST_ASSERT_EQUAL_INT(0,
        xce_hook_init(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_MAX));

    g_tace_c.jobmgr    = makeInstance(&xkvmgr_type, &int_params);
    g_tace_c.qmgr      = makeInstance(&xkvmgr_type, &str_params);
    g_tace_c.workermgr = makeInstance(&xkvmgr_type, &str_params);
    g_tace_c.resmgr    = makeInstance(&xkvmgr_type, &str_params);
    TEST_ASSERT_NOT_NULL(g_tace_c.jobmgr);
    TEST_ASSERT_NOT_NULL(g_tace_c.qmgr);
    TEST_ASSERT_NOT_NULL(g_tace_c.workermgr);
    TEST_ASSERT_NOT_NULL(g_tace_c.resmgr);
    zarray_init(&g_tace_c.sharedinst, 16, NULL);

    memset(&g_tace_p, 0, sizeof(g_tace_p));
    g_tace_p.cluster = &g_tace_c;

    g_tace_hook_fires = 0;
}

static void tace_teardown(void)
{
    zarray_fini(&g_tace_c.sharedinst);
    if (g_tace_c.resmgr)    dereference(g_tace_c.resmgr);
    if (g_tace_c.workermgr) dereference(g_tace_c.workermgr);
    if (g_tace_c.qmgr)      dereference(g_tace_c.qmgr);
    if (g_tace_c.jobmgr)    dereference(g_tace_c.jobmgr);
    xce_hook_fini(&g_tace_c.hooksys);
}

/* ── Encode helpers (mirror the static capnp_encode_{job,worker} in
 *    persist.c — those are file-static, so tests keep local copies, exactly
 *    as test_egress_filter.c's ef_encode_job already does). ────────────── */

static int tace_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;
}

static int tace_encode_worker(const xce_worker_t* w, uint8_t** out,
                              size_t* out_len)
{
    struct capn          c;
    struct capn_segment* cs;
    struct Worker        cw;
    Worker_ptr           wp;

    capn_init_malloc(&c);
    cs = capn_root(&c).seg;
    encode_Worker(cs, &cw, (xce_worker_t*)w);
    wp = new_Worker(cs);
    write_Worker(&cw, wp);
    capn_setp(capn_root(&c), 0, wp.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;
}

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

/* Creates a job directly in the fixture's jobmgr, bypassing
 * handle_jobsubmit/restore_job entirely — for seeding fact-transition tests. */
static job_submit_data_t g_tace_seed_submit;

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

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

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

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

/* ── WAL_JOB_SUBMIT ───────────────────────────────────────────────────── */

static void test_ace_job_submit_constructs_and_notifies(void)
{
    tace_setup();

    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  = 501;
    jdef.state  = JOB_STAT_PEND;
    jdef.submit = &submit;

    uint8_t* payload = NULL;
    size_t   plen    = 0;
    TEST_ASSERT_EQUAL_INT(0, tace_encode_job(&jdef, &payload, &plen));

    TEST_ASSERT_EQUAL_INT(0,
        xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB,
                     tace_spy_hook, NULL));

    int rc = apply_committed_entry(&g_tace_p, WAL_JOB_SUBMIT, 10, payload,
                                   (uint32_t)plen);
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, rc,
        "a genuine new job creation must return 1 (bug-1419: distinguishes "
        "creation from an idempotent duplicate no-op)");

    job_t* j = tace_find_job(501);
    TEST_ASSERT_NOT_NULL_MESSAGE(j,
        "WAL_JOB_SUBMIT must construct the job via restore_job");
    TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state);
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
        "a fresh job submit must fire exactly one post-mutation notification");

    free(payload);
    tace_teardown();
    printf("test_ace_job_submit_constructs_and_notifies: PASS\n");
}

static void test_ace_job_submit_duplicate_is_noop_no_extra_notify(void)
{
    tace_setup();

    xce_job_t jdef;
    job_submit_data_t submit;
    memset(&submit, 0, sizeof(submit));
    submit.queue = (char*)"q";
    submit.user  = (char*)"tester";
    memset(&jdef, 0, sizeof(jdef));
    jdef.jobid  = 502;
    jdef.state  = JOB_STAT_PEND;
    jdef.submit = &submit;

    uint8_t* payload = NULL;
    size_t   plen    = 0;
    TEST_ASSERT_EQUAL_INT(0, tace_encode_job(&jdef, &payload, &plen));

    xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tace_spy_hook,
                NULL);

    TEST_ASSERT_EQUAL_INT_MESSAGE(1, apply_committed_entry(&g_tace_p,
                                                            WAL_JOB_SUBMIT, 11,
                                                            payload,
                                                            (uint32_t)plen),
        "a genuine new job creation must return 1");
    TEST_ASSERT_EQUAL_INT(1, g_tace_hook_fires);

    /* Re-apply the SAME entry (replay / resync redelivery) — must be a
     * no-op: no error, no second job, no second notification (Req 7.1). */
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, apply_committed_entry(&g_tace_p,
                                                            WAL_JOB_SUBMIT, 11,
                                                            payload,
                                                            (uint32_t)plen),
        "an idempotent duplicate must return 0, not 1");
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
        "an idempotent duplicate submit must not fire a second notification");

    free(payload);
    tace_teardown();
    printf("test_ace_job_submit_duplicate_is_noop_no_extra_notify: PASS\n");
}

/* ── WAL_JOB_START / job-state fact family ───────────────────────────── */

static void test_ace_job_state_transition_applies_forward_and_notifies(void)
{
    tace_setup();

    job_t* j = tace_seed_job(601, JOB_STAT_PEND);
    TEST_ASSERT_NOT_NULL(j);

    wal_jobstate_t js = { .jobid = 601, .state = JOB_STAT_RUN };

    xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tace_spy_hook,
                NULL);

    int rc = apply_committed_entry(&g_tace_p, WAL_JOB_START, 20, &js,
                                   sizeof(js));
    TEST_ASSERT_EQUAL_INT(0, rc);
    TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state,
        "a forward job-state fact must apply via the 2.1 guarded core");
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
        "a forward job-state transition must fire exactly one notification");

    tace_teardown();
    printf("test_ace_job_state_transition_applies_forward_and_notifies: "
          "PASS\n");
}

static void
test_ace_job_state_transition_rejects_backward_suppresses_notify(void)
{
    tace_setup();

    /* Job already terminal (as if FINISH already applied); a stale,
     * previously-unapplied START must not regress it (Req 7.1/7.2). */
    job_t* j = tace_seed_job(602, JOB_STAT_DONE);
    TEST_ASSERT_NOT_NULL(j);

    wal_jobstate_t js = { .jobid = 602, .state = JOB_STAT_RUN };

    xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tace_spy_hook,
                NULL);

    int rc = apply_committed_entry(&g_tace_p, WAL_JOB_START, 21, &js,
                                   sizeof(js));
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
        "a guard-rejected backward transition is a success no-op, not an "
        "apply failure");
    TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j->job.state,
        "a rejected backward transition must preserve the terminal state");
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_tace_hook_fires,
        "a guard-rejected transition must fire NO notification");

    tace_teardown();
    printf("test_ace_job_state_transition_rejects_backward_suppresses_"
          "notify: PASS\n");
}

/* ── WAL_QUEUE_CTRL ───────────────────────────────────────────────────── */

static void test_ace_queue_ctrl_updates_and_notifies(void)
{
    tace_setup();

    xce_queue_t qdef;
    memset(&qdef, 0, sizeof(qdef));
    qdef.queue = (char*)"q1";
    TEST_ASSERT_NOT_NULL(makeQueue(g_tace_c.qmgr, &qdef));

    const char* name     = "q1";
    size_t      name_len = strlen(name) + 1;
    size_t      total    = sizeof(wal_queuehdr_t) + name_len;
    uint8_t*    pkt       = malloc(total);
    TEST_ASSERT_NOT_NULL(pkt);
    wal_queuehdr_t qh = { .status = QUEUE_STAT_CLOSED };
    memcpy(pkt, &qh, sizeof(qh));
    memcpy(pkt + sizeof(qh), name, name_len);

    xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_QUEUE, tace_spy_hook,
                NULL);

    int rc = apply_committed_entry(&g_tace_p, WAL_QUEUE_CTRL, 30, pkt,
                                   (uint32_t)total);
    TEST_ASSERT_EQUAL_INT(0, rc);

    queue_t* q = NULL;
    lock(g_tace_c.qmgr, RTTI_LOCK_READ);
    key_find(g_tace_c.qmgr, (void*)name, (void**)&q);
    unlock(g_tace_c.qmgr);
    TEST_ASSERT_NOT_NULL(q);
    TEST_ASSERT_TRUE_MESSAGE((q->queue.status & QUEUE_STAT_CLOSED) != 0,
        "WAL_QUEUE_CTRL must apply via restore_queue_state");
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
        "a queue control apply must fire exactly one notification");

    free(pkt);
    tace_teardown();
    printf("test_ace_queue_ctrl_updates_and_notifies: PASS\n");
}

/* ── WAL_WORKER_CTRL ──────────────────────────────────────────────────── */

static void test_ace_worker_ctrl_join_restores_and_notifies(void)
{
    tace_setup();

    /* copy_worker (wobj.c) rejects a NULL src->attrs with -EINVAL — a real
     * WAL_WORKER_CTRL JOIN payload always carries real host attributes, so
     * the fixture must too. */
    hostattrs_t attrs;
    memset(&attrs, 0, sizeof(attrs));
    attrs.hosttype  = (char*)"unknown";
    attrs.hostmodel = (char*)"unknown";

    xce_worker_t wdef;
    memset(&wdef, 0, sizeof(wdef));
    wdef.name  = (char*)"w1";
    wdef.attrs = &attrs;

    uint8_t* wbuf  = NULL;
    size_t   wsize = 0;
    TEST_ASSERT_EQUAL_INT(0, tace_encode_worker(&wdef, &wbuf, &wsize));

    uint32_t disc  = (uint32_t)HOOK_DETAIL_WORKER_JOIN;
    size_t   total = sizeof(disc) + wsize;
    uint8_t* pkt   = malloc(total);
    TEST_ASSERT_NOT_NULL(pkt);
    memcpy(pkt, &disc, sizeof(disc));
    memcpy(pkt + sizeof(disc), wbuf, wsize);
    free(wbuf);

    xce_add_hook(&g_tace_c.hooksys, XCE_CONTROLLER_HOOK_WORKER, tace_spy_hook,
                NULL);

    int rc = apply_committed_entry(&g_tace_p, WAL_WORKER_CTRL, 40, pkt,
                                   (uint32_t)total);
    TEST_ASSERT_EQUAL_INT(0, rc);

    worker_t* w = NULL;
    lock(g_tace_c.workermgr, RTTI_LOCK_READ);
    key_find(g_tace_c.workermgr, (void*)"w1", (void**)&w);
    unlock(g_tace_c.workermgr);
    TEST_ASSERT_NOT_NULL_MESSAGE(w,
        "WAL_WORKER_CTRL must restore the worker via worker_from_capnp");
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_tace_hook_fires,
        "a worker control apply must fire exactly one notification");

    free(pkt);
    tace_teardown();
    printf("test_ace_worker_ctrl_join_restores_and_notifies: PASS\n");
}

/* ── Defensive / boundary ─────────────────────────────────────────────── */

static void test_ace_null_persist_returns_error(void)
{
    int rc = apply_committed_entry(NULL, WAL_JOB_UPDATE, 1, NULL, 0);
    TEST_ASSERT_NOT_EQUAL(0, rc);
    printf("test_ace_null_persist_returns_error: PASS\n");
}

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

void test_apply_committed_entry(void)
{
    UNITY_BEGIN();
    RUN_TEST(test_ace_job_submit_constructs_and_notifies);
    RUN_TEST(test_ace_job_submit_duplicate_is_noop_no_extra_notify);
    RUN_TEST(test_ace_job_state_transition_applies_forward_and_notifies);
    RUN_TEST(test_ace_job_state_transition_rejects_backward_suppresses_notify);
    RUN_TEST(test_ace_queue_ctrl_updates_and_notifies);
    RUN_TEST(test_ace_worker_ctrl_join_restores_and_notifies);
    RUN_TEST(test_ace_null_persist_returns_error);
    UNITY_END();
}