myxce/src/controller/testsuite/test_replay_apply_seam.c

/* test_replay_apply_seam.c — raft-apply-decoupling task 2.3
 *
 * Proves restart replay (persist_recover -> replay_wal) applies every
 * classified WAL entry type through the SAME apply_committed_entry() seam
 * task 2.2 built (routing job submit through restore_job, job-state facts
 * through the 2.1 guarded core, queue control through restore_queue_state,
 * worker control through worker_from_capnp) instead of the pre-2.3 replay
 * path's direct calls inside apply_wal_entry().
 *
 * Two things this proves:
 *
 *   1. Materialized-state parity (Req 3.1/3.2): a log containing one entry
 *      of every classified type — WAL_JOB_SUBMIT, a job-state fact
 *      (WAL_JOB_START), WAL_QUEUE_CTRL, WAL_WORKER_CTRL — replayed via
 *      persist_recover() on a fresh persist_t/cluster_t produces the
 *      correct end state. For this causally-ordered (submit-before-start),
 *      non-duplicate log, this is identical to what the pre-task-2.3
 *      apply_wal_entry() dispatch already produced: neither switch body
 *      differs in materialized outcome for a forward, single-instance
 *      apply. This is the "pure refactor, existing behavior preserved"
 *      evidence for the call-site swap.
 *
 *   2. The call site genuinely switched, not just "happens to look the
 *      same": unlike apply_wal_entry() (which never fires a notification
 *      hook), apply_committed_entry() fires each type's AFTER_EVENT hook
 *      exactly once, strictly after its mutation. A spy hook counts these
 *      firings across the whole replay. Before task 2.3 (replay wired to
 *      apply_wal_entry) this count is 0; after (replay wired to
 *      apply_committed_entry) it is exactly 4 — one per entry. That
 *      flip is genuine RED (pre-change) / GREEN (post-change) evidence
 *      that replay's per-entry apply call site was actually rewired.
 *
 * Fixture mirrors test_apply_committed_entry.c's cluster_t construction
 * (task 2.2), adapted to drive the real physical write path
 * (persist_wal_async) and the real recovery path (persist_recover) instead
 * of calling the dispatch function directly.
 */
#include <rtti.h>

#include <capnp_c.h>
#include <errno.h>
#include <inttypes.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 "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 / test_apply_committed_entry.c. */
static int g_trr_jobobj_ready;

static cluster_t g_trr_c;

/* ── Spy hook: counts AFTER_EVENT firings across job/queue/worker hooks ─── */
static int g_trr_hook_fires;

static int trr_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_trr_hook_fires++;
    return 0;
}

static void trr_build_cluster(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*)"trr-probe") == NULL)
        xstring_table_init(64);

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

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

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

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

    /* Queues come from config, not from WAL_QUEUE_CTRL (restore_queue_state
     * -ENOENT's "queue not yet built from config" is a benign replay skip)
     * — pre-create the queue the replayed WAL_QUEUE_CTRL entry targets,
     * exactly as a real boot's config load would before replay runs. */
    xce_queue_t qdef;
    memset(&qdef, 0, sizeof(qdef));
    qdef.queue = (char*)"rq1";
    TEST_ASSERT_NOT_NULL(makeQueue(g_trr_c.qmgr, &qdef));

    xce_add_hook(&g_trr_c.hooksys, XCE_CONTROLLER_HOOK_JOB, trr_spy_hook,
                NULL);
    xce_add_hook(&g_trr_c.hooksys, XCE_CONTROLLER_HOOK_QUEUE, trr_spy_hook,
                NULL);
    xce_add_hook(&g_trr_c.hooksys, XCE_CONTROLLER_HOOK_WORKER, trr_spy_hook,
                NULL);

    g_trr_hook_fires = 0;
}

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

/* ── Encode helpers (mirror test_apply_committed_entry.c's tace_encode_*;
 *    capnp_encode_{job,worker} are file-static in persist.c). ──────────── */

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

/* ── persist_t fixture (real disk-backed WAL, mirrors
 *    test_persist_recovery.c's make_fresh_persist). ─────────────────────── */

static persist_t* trr_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 */
    cfg.snap_keep_n = 2;

    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 trr_teardown_persist(persist_t* p)
{
    persist_stop(p);
    persist_fini(p);
    free(p);
}

/* Look up helpers, mirroring test_apply_committed_entry.c. */
static job_t* trr_find_job(uint64_t jobid)
{
    job_t* j = NULL;
    lock(g_trr_c.jobmgr, RTTI_LOCK_READ);
    int found = key_find(g_trr_c.jobmgr, (void*)jobid, (void**)&j);
    unlock(g_trr_c.jobmgr);
    return (found > 0) ? j : NULL;
}

static queue_t* trr_find_queue(const char* name)
{
    queue_t* q = NULL;
    lock(g_trr_c.qmgr, RTTI_LOCK_READ);
    key_find(g_trr_c.qmgr, (void*)name, (void**)&q);
    unlock(g_trr_c.qmgr);
    return q;
}

static worker_t* trr_find_worker(const char* name)
{
    worker_t* w = NULL;
    lock(g_trr_c.workermgr, RTTI_LOCK_READ);
    key_find(g_trr_c.workermgr, (void*)name, (void**)&w);
    unlock(g_trr_c.workermgr);
    return w;
}

/*
 * Writes one physical WAL entry of every classified type (real
 * persist_wal_async appends — no hand-rolled headers/CRCs) in causal order:
 * submit before the state fact that mutates the same job.
 */
static void trr_write_classified_log(persist_t* p)
{
    /* WAL_JOB_SUBMIT: jobid 90001, PEND */
    job_submit_data_t submit;
    memset(&submit, 0, sizeof(submit));
    submit.queue = (char*)"rq1";
    submit.user  = (char*)"tester";

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

    uint8_t* job_payload = NULL;
    size_t   job_plen    = 0;
    TEST_ASSERT_EQUAL_INT(0, trr_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);

    /* WAL_JOB_START (job-state fact family): jobid 90001 -> RUN */
    wal_jobstate_t js = { .jobid = 90001, .state = JOB_STAT_RUN };
    TEST_ASSERT_EQUAL_INT(0,
        persist_wal_async(p, WAL_JOB_START, &js, sizeof(js)));

    /* WAL_QUEUE_CTRL: close queue "rq1" */
    const char* qname     = "rq1";
    size_t      qname_len = strlen(qname) + 1;
    size_t      qtotal    = sizeof(wal_queuehdr_t) + qname_len;
    uint8_t*    qpkt      = malloc(qtotal);
    TEST_ASSERT_NOT_NULL(qpkt);
    wal_queuehdr_t qh = { .status = QUEUE_STAT_CLOSED };
    memcpy(qpkt, &qh, sizeof(qh));
    memcpy(qpkt + sizeof(qh), qname, qname_len);
    TEST_ASSERT_EQUAL_INT(0,
        persist_wal_async(p, WAL_QUEUE_CTRL, qpkt, (uint32_t)qtotal));
    free(qpkt);

    /* WAL_WORKER_CTRL: join worker "rw1" */
    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*)"rw1";
    wdef.attrs = &attrs;

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

    uint32_t disc  = (uint32_t)HOOK_DETAIL_WORKER_JOIN;
    size_t   wtotal = sizeof(disc) + wsize;
    uint8_t* wpkt   = malloc(wtotal);
    TEST_ASSERT_NOT_NULL(wpkt);
    memcpy(wpkt, &disc, sizeof(disc));
    memcpy(wpkt + sizeof(disc), wbuf, wsize);
    free(wbuf);
    TEST_ASSERT_EQUAL_INT(0,
        persist_wal_async(p, WAL_WORKER_CTRL, wpkt, (uint32_t)wtotal));
    free(wpkt);

    TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
}

/*
 * Req 3.1/3.2 + genuine-rewire evidence: restart replay of a log containing
 * every classified entry type reaches the correct end state, and does so by
 * actually firing apply_committed_entry()'s post-mutation notification for
 * each entry (proving replay is wired to the new seam, not merely producing
 * a coincidentally-matching state).
 */
static void test_replay_applies_every_classified_type_via_apply_seam(void)
{
    char tmp[256];
    strcpy(tmp, "/tmp/xce_replay_seam_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(tmp));

    /* Phase 1: write one entry of every classified type to a real WAL. No
     * cluster needed — persist_wal_async is a physical-write-only path. */
    {
        persist_t* p = trr_make_fresh_persist(tmp);
        TEST_ASSERT_NOT_NULL(p);
        trr_write_classified_log(p);
        trr_teardown_persist(p);
    }

    /* Phase 2: fresh persist_t + fresh cluster_t, recover from disk. */
    {
        persist_t* p2 = trr_make_fresh_persist(tmp);
        TEST_ASSERT_NOT_NULL(p2);

        trr_build_cluster();
        p2->cluster = &g_trr_c;

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

        job_t* j = trr_find_job(90001);
        TEST_ASSERT_NOT_NULL_MESSAGE(j,
            "replay must construct the job via the apply seam's "
            "WAL_JOB_SUBMIT route");
        TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j->job.state,
            "replay must apply the job-state fact via the apply seam's "
            "guarded transition core, reaching the same end state as before");

        queue_t* q = trr_find_queue("rq1");
        TEST_ASSERT_NOT_NULL(q);
        TEST_ASSERT_TRUE_MESSAGE((q->queue.status & QUEUE_STAT_CLOSED) != 0,
            "replay must apply WAL_QUEUE_CTRL via the apply seam");

        worker_t* w = trr_find_worker("rw1");
        TEST_ASSERT_NOT_NULL_MESSAGE(w,
            "replay must apply WAL_WORKER_CTRL via the apply seam");

        /* One notification per entry: JOB_NEW, JOB_UPDATE, QUEUE_UPDATE,
         * WORKER_JOIN. apply_wal_entry (the pre-2.3 replay dispatch) never
         * fires hooks — this assertion is 0 there and 4 once replay is
         * pointed at apply_committed_entry, proving the call site moved. */
        TEST_ASSERT_EQUAL_INT_MESSAGE(4, g_trr_hook_fires,
            "restart replay must drive every classified entry through "
            "apply_committed_entry(), firing exactly one post-mutation "
            "notification per entry");

        trr_teardown_cluster();
        trr_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_replay_applies_every_classified_type_via_apply_seam: PASS\n");
}

void test_replay_apply_seam(void)
{
    RUN_TEST(test_replay_applies_every_classified_type_via_apply_seam);
}