myxce/src/controller/testsuite/test_jobdispatch_no_double_notify.c

/* test_jobdispatch_no_double_notify.c -- scheduler-dispatch-commit task 4.1
 *
 * Proves handle_jobdispatch() (job.c) no longer duplicates the dispatch-
 * outcome reporting that dispatch_a_job_() (tasks 3.1-3.3) now owns
 * entirely. Before this task, handle_jobdispatch()'s loop called
 * notify_dispatch() itself for EVERY job in the batch regardless of
 * dispatch_a_job_()'s own outcome/classification -- on top of whatever
 * dispatch_a_job_() had already decided to do (nothing on success or on a
 * terminal failure, exactly one notify_dispatch() call on a retryable
 * failure). That produced a conflicting double report for a terminal
 * failure specifically (a caller-added jobDispatchFail alongside
 * mark_job_fail()'s own schedFailJob -- design.md's round-2 finding) and a
 * redundant extra notify_dispatch() call for a retryable failure.
 *
 * This test drives handle_jobdispatch() directly (not dispatch_a_job_test())
 * with a 3-job batch covering all three of dispatch_a_job_()'s own
 * notify_dispatch() behaviors:
 *   - success (a live, ready worker)          -> dispatch_a_job_() itself
 *     calls notify_dispatch() 0 times
 *   - retryable failure (unregistered worker) -> dispatch_a_job_() itself
 *     calls notify_dispatch() exactly 1 time (below the retry bound)
 *   - terminal failure (nhosts == 0)          -> dispatch_a_job_() itself
 *     calls notify_dispatch() 0 times (mark_job_fail()'s hook chain is the
 *     sole notification for this case)
 *
 * dispatch_a_job_() alone (already proven correct by tasks 3.1-3.3's own
 * test_dispatch_commit.c / test_dispatch_fail_terminal.c /
 * test_dispatch_fail_retry.c) therefore produces exactly ONE
 * notify_dispatch() call for this batch. The fix under test is that
 * handle_jobdispatch() must not add anything on top of that -- the total
 * observed via notify() (test_commit_barrier_reply.c's capture stub, the
 * sole notify() definition linked into this binary) must be exactly 1, not
 * 4 (1 extra per job x 3, on top of the 1 dispatch_a_job_() already
 * produces) as it was before this task.
 *
 * Fixture mirrors test_dispatch_commit.c's dc_setup/dc_teardown (cluster_t
 * with a real disk-backed persist_t, cluster == NULL routing so
 * persist_submit_job_fact() takes the direct persist_wal_async() path, no
 * HA/commit-engine machinery needed) plus test_commit_barrier_reply.c's
 * g_req_ep / c->myhost requester-endpoint pattern (zeroed, no iomgr --
 * xce_reply()'s enqueue always succeeds against it, proven by that file's
 * own header) so handle_jobdispatch()'s receipt-ACK xce_reply() call has
 * somewhere to enqueue to.
 *
 * notify_dispatch() observability: see test_dispatch_fail_retry.c's header
 * for the full rationale -- this binary's sole notify() definition lives in
 * test_commit_barrier_reply.c, exposed read-only via its
 * cbrp_notify_capture_*_ext() accessors (added by task 3.3).
 *
 * scheduler-dispatch-commit task 5.3 (Req 4.2, 4.5) adds
 * test_handle_jobdispatch_terminal_no_notify_no_leak, below. It is a
 * sharper-focused sibling of this file's original (task 4.1) 3-job-batch
 * test: that test proves the BATCH's total notify_dispatch() count is 1,
 * attributable to the retryable job (an aggregate proof the terminal job
 * contributed zero calls only by subtraction/arithmetic against the other
 * two jobs' known outcomes) and never exercises reference-count cleanliness
 * for any job. The new test instead drives a SINGLE terminal-classified job
 * through this same real handle_jobdispatch() entry point (not
 * dispatch_a_job_test()'s bypass, unlike test_dispatch_fail_terminal.c's
 * task-3.2 tests) and combines, in ONE test run: (a) JOB_STAT_EXIT
 * resolution, (b) a direct (not subtracted) zero-notify assertion for this
 * job's own batch, and (c) test_dispatch_fail_terminal.c's canary-reference
 * technique proving the reference count returns to its pre-attempt value.
 */

#include <rtti.h>

#include <errno.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 "dataq.h"
#include "hook.h"
#include "jobj.h"
#include "kvmgr.h"
#include "netio.h"
#include "persist.h"
#include "qobj.h"
#include "typemap.h"
#include "wobj.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xcealloc.h"
#include "xstring.h"

/* test_commit_barrier_reply.c's minimal notify()-capture accessors (task
 * 3.3) -- see this file's header. This binary's sole notify() definition
 * lives there (dataex.c's real one is excluded from the unit-test link). */
extern int      cbrp_notify_capture_count_ext(void);
extern int      cbrp_notify_capture_event_ext(int idx);
extern uint64_t cbrp_notify_capture_jobid_ext(int idx);
extern int      cbrp_notify_capture_code_ext(int idx);
extern void     cbrp_notify_capture_reset_ext(void);

/* jobj.c's job type 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_jnd_jobobj_ready;

static cluster_t      g_jnd;
static char           g_jnd_tmpdir[256];
static endpoint_t     g_jnd_req_ep;    /* requester endpoint (receipt ACK) */
static endpoint_t     g_jnd_worker_ep; /* the one live worker's endpoint   */
static xce_peerinfo_t g_jnd_req_from;

static void jnd_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 };
    persist_config_t cfg;

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

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

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

    g_jnd.jobmgr    = makeInstance(&xkvmgr_type, &int_params);
    g_jnd.workermgr = makeInstance(&xkvmgr_type, &str_params);
    g_jnd.qmgr      = makeInstance(&xkvmgr_type, &str_params);
    TEST_ASSERT_NOT_NULL(g_jnd.jobmgr);
    TEST_ASSERT_NOT_NULL(g_jnd.workermgr);
    TEST_ASSERT_NOT_NULL(g_jnd.qmgr);

    /* Real disk-backed persist, cluster == NULL: wal_hook_append_
     * classified() takes the direct persist_wal_async() path -- no raft/
     * commit-engine machinery needed for the successful job's plain
     * WAL_JOB_DISPATCH fact append (mirrors test_dispatch_commit.c). */
    cfg             = persist_config_defaults();
    cfg.wal_sync_ms = 5000;
    strcpy(g_jnd_tmpdir, "/tmp/xce_jnd41_XXXXXX");
    TEST_ASSERT_NOT_NULL(mkdtemp(g_jnd_tmpdir));
    persist_set_workdir(&g_jnd.persist, g_jnd_tmpdir, "ctrl-test");
    TEST_ASSERT_EQUAL_INT(0, persist_init(&g_jnd.persist, &cfg, NULL));
    TEST_ASSERT_EQUAL_INT(0, persist_start(&g_jnd.persist));

    g_jnd.myhost           = &g_jnd.myhost_info;
    g_jnd.myhost_info.name = (char*)"ctrl-test";
    g_jnd.myhost_info.key  = (char*)"";

    /* Zeroed, no iomgr -- xce_reply()'s/xce_startjob_req()'s enqueue always
     * succeeds against an endpoint like this (test_dispatch_commit.c's and
     * test_commit_barrier_reply.c's own established, proven pattern). */
    memset(&g_jnd_req_ep, 0, sizeof(g_jnd_req_ep));
    g_jnd_req_ep.fd = -1;

    memset(&g_jnd_worker_ep, 0, sizeof(g_jnd_worker_ep));
    g_jnd_worker_ep.fd = -1;

    g_jnd_req_from.name = (char*)"requester";
    g_jnd_req_from.key  = (char*)"";
}

/* Frees every packet left queued on an endpoint's outq -- mirrors
 * test_dispatch_commit.c's dc_drain_ep. */
static void jnd_drain_ep(endpoint_t* ep)
{
    data_item_t* it;

    while ((it = endpoint_deq(ep)) != NULL) {
        io_packet_t* pkt = ITPTR_TO_PACKET(it);
        free_packet(pkt);
        free(pkt);
    }
}

static void jnd_teardown(void)
{
    char cmd[512];

    jnd_drain_ep(&g_jnd_req_ep);
    jnd_drain_ep(&g_jnd_worker_ep);

    persist_stop(&g_jnd.persist);
    persist_fini(&g_jnd.persist);

    if (g_jnd.workermgr) dereference(g_jnd.workermgr);
    if (g_jnd.qmgr)      dereference(g_jnd.qmgr);
    if (g_jnd.jobmgr)    dereference(g_jnd.jobmgr);
    xce_hook_fini(&g_jnd.hooksys);

    snprintf(cmd, sizeof(cmd), "rm -rf %s", g_jnd_tmpdir);
    if (system(cmd) != 0)
        fprintf(stderr, "cleanup warning: rm -rf %s failed\n", g_jnd_tmpdir);
}

/* Seeds a queue and a PEND job with a real submit/start_info against
 * g_jnd.jobmgr/qmgr (dispatch_a_job_() reads submit->queue and
 * submit->start_info). Returns the seeded job_t* (borrowed -- g_jnd.jobmgr
 * owns the creation reference). Mirrors test_dispatch_fail_retry.c's
 * dfr_seed_job. */
static job_t* jnd_seed_job(uint64_t jobid, const char* qname)
{
    xce_queue_t       qdef;
    job_startinfo_t   startinfo;
    job_submit_data_t submit;
    xce_job_t         jdef;
    job_t*            j;

    memset(&qdef, 0, sizeof(qdef));
    qdef.queue = (char*)qname;
    TEST_ASSERT_NOT_NULL(makeQueue(g_jnd.qmgr, &qdef));

    memset(&startinfo, 0, sizeof(startinfo));
    startinfo.cmdline = (char*)"/bin/true";
    memset(&submit, 0, sizeof(submit));
    submit.queue      = (char*)qname;
    submit.user       = (char*)"tester";
    submit.start_info = &startinfo;

    memset(&jdef, 0, sizeof(jdef));
    jdef.jobid  = jobid;
    jdef.state  = JOB_STAT_PEND;
    jdef.submit = &submit;
    j = makeJob(g_jnd.jobmgr, &jdef);
    TEST_ASSERT_NOT_NULL(j);

    return j;
}

/* Registers a ready worker (host.status defaults to 0 via calloc) in
 * g_jnd.workermgr with ep pointed at the given endpoint. Mirrors
 * test_dispatch_commit.c's dc_seed_ready's worker half. */
static void jnd_seed_ready_worker(const char* wname, endpoint_t* ep)
{
    xce_worker_t wdef;
    hostattrs_t* attrs;
    worker_t*    w = NULL;

    attrs = calloc(1, sizeof(*attrs));
    TEST_ASSERT_NOT_NULL(attrs);
    attrs->hosttype  = makeString("LINUX");
    attrs->hostmodel = makeString("x86");
    attrs->cpufactor = 1.0f;
    attrs->maxcpus   = 4;
    memset(&wdef, 0, sizeof(wdef));
    wdef.name  = (char*)wname;
    wdef.attrs = attrs;
    TEST_ASSERT_NOT_NULL(makeWorker(g_jnd.workermgr, &wdef));

    lock(g_jnd.workermgr, RTTI_LOCK_READ);
    key_find(g_jnd.workermgr, (void*)wname, (void**)&w);
    unlock(g_jnd.workermgr);
    TEST_ASSERT_NOT_NULL(w);
    w->ep = ep;
}

#define JND_JOBID_SUCCESS   92101ULL
#define JND_JOBID_RETRYABLE 92102ULL
#define JND_JOBID_TERMINAL  92103ULL

/* THE core proof (Req 3.1, 4.1, 4.2): a batch mixing all three of
 * dispatch_a_job_()'s own notify_dispatch() behaviors must produce EXACTLY
 * what dispatch_a_job_() itself produces (1 -- the retryable job's own
 * call), not 1 extra per job from handle_jobdispatch()'s own loop. */
static void test_handle_jobdispatch_matches_dispatch_a_job_notify_count(void)
{
    xce_job_alloc_t*   allocs[3];
    xce_job_dispatch_t dispatchmsg;
    xce_request_t      req;
    job_t*              jsucc;
    job_t*              jretry;
    job_t*              jterm;
    int                rc;

    jnd_setup();

    jsucc  = jnd_seed_job(JND_JOBID_SUCCESS, "jnd-q1");
    jretry = jnd_seed_job(JND_JOBID_RETRYABLE, "jnd-q2");
    jterm  = jnd_seed_job(JND_JOBID_TERMINAL, "jnd-q3");

    jnd_seed_ready_worker("jnd-wkr-ok", &g_jnd_worker_ep);
    /* jretry deliberately targets an unregistered worker name -- no
     * corresponding makeWorker() call, matching test_dispatch_fail_retry.c's
     * own established WORKER_NULL/WORKER_LOOKUP fixture. */

    allocs[0] = make_job_alloc(JND_JOBID_SUCCESS, 1,
        make_host_allocs(1,
            make_host_alloc((char*)"jnd-wkr-ok", 1,
                make_allocs(1,
                    make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
    TEST_ASSERT_NOT_NULL(allocs[0]);

    allocs[1] = make_job_alloc(JND_JOBID_RETRYABLE, 1,
        make_host_allocs(1,
            make_host_alloc((char*)"jnd-no-such-worker", 1,
                make_allocs(1,
                    make_alloc((char*)"slots", NULL, 0, 1.0f, 0)))));
    TEST_ASSERT_NOT_NULL(allocs[1]);

    /* nhosts == 0: XCE_DISPATCH_FAIL_SITE_INVALID_ALLOC, terminal --
     * mirrors test_commit_barrier_reply.c's cbrp_arrange_jobdispatch()
     * second allocation exactly. */
    allocs[2] = make_job_alloc(JND_JOBID_TERMINAL, 0, NULL);
    TEST_ASSERT_NOT_NULL(allocs[2]);

    memset(&dispatchmsg, 0, sizeof(dispatchmsg));
    dispatchmsg.njobs  = 3;
    dispatchmsg.allocs = allocs;

    memset(&req, 0, sizeof(req));
    req.opcode           = OPCode_jobdispatch;
    req.seq              = 42;
    req.from             = &g_jnd_req_from;
    req.body_which       = Request_body_dispatchmsg;
    req.body.dispatchmsg = &dispatchmsg;

    cbrp_notify_capture_reset_ext();

    /* Act: handle_jobdispatch() consumes/frees every alloc itself via
     * MOVE_PTR + free_JobAlloc_ptr() in its own loop -- nothing left for
     * this test to free afterward. */
    rc = handle_jobdispatch(&g_jnd, &g_jnd_req_ep, &req);
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "handle_jobdispatch's own return "
        "is unconditionally 0, unaffected by any per-job outcome");

    TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, jsucc->job.state,
        "the successful dispatch must still commit RUN (unaffected by this "
        "task -- only the caller's own notification logic changes)");
    TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, jretry->job.state,
        "a below-bound retryable failure must still leave the job PEND");
    TEST_ASSERT_EQUAL_UINT32_MESSAGE(1, jretry->dispatch_fail_count,
        "the retryable failure's own counter increment is unaffected by "
        "this task");
    TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_EXIT, jterm->job.state,
        "the terminal (nhosts == 0) failure must still resolve to EXIT via "
        "mark_job_fail(), unaffected by this task");

    /* THE assertion this task exists for: dispatch_a_job_() alone (tasks
     * 3.1-3.3, already proven correct) produces exactly ONE
     * notify_dispatch() call for this exact batch -- the retryable job's
     * own call. Before this task, handle_jobdispatch()'s own unconditional
     * per-job notify_dispatch() calls would add one more per job (3), for a
     * total of 4 -- a conflicting double (and, for the terminal job,
     * mismatched) report. */
    TEST_ASSERT_EQUAL_INT_MESSAGE(1, cbrp_notify_capture_count_ext(),
        "handle_jobdispatch() must not add any notify_dispatch() call on "
        "top of what dispatch_a_job_() itself already produced for this "
        "batch (Req 3.1, 4.1, 4.2) -- exactly the retryable job's own "
        "call, none for the success or terminal jobs");
    TEST_ASSERT_EQUAL_INT(NotifyData_data_jobDispatchFail,
        cbrp_notify_capture_event_ext(0));
    TEST_ASSERT_EQUAL_UINT64(JND_JOBID_RETRYABLE,
        cbrp_notify_capture_jobid_ext(0));

    jnd_teardown();
    printf("test_handle_jobdispatch_matches_dispatch_a_job_notify_count: "
          "PASS\n");
}

#define JND_JOBID_TERMINAL_SOLO 92201ULL

/* Canary-reference technique, reused verbatim from
 * test_dispatch_fail_terminal.c's dft_arm_ref_canary()/dft_assert_ref_
 * not_over_released() (task 3.2): armed immediately BEFORE the call under
 * test (not after -- arming after would rely incidentally on other
 * assertions crashing first, which doesn't independently prove anything),
 * released and the job re-touched immediately after. An over-release during
 * the call would free j while the canary reference is still held, making
 * the post-call touch a real, ASAN-caught heap-use-after-free. job_t's real
 * refcount lives in jobj.c's file-local job_private_t with no accessor
 * exposed to a test, so this indirect technique is the only available
 * proof. */
static void jnd_arm_ref_canary(job_t* j)
{
    reference(j);
}

static void jnd_assert_ref_not_over_released(job_t* j, uint64_t expect_jobid)
{
    dereference(j);

    TEST_ASSERT_EQUAL_UINT64_MESSAGE(expect_jobid, j->job.jobid,
        "Req 4.5: j must still be safely readable after releasing the "
        "canary reference -- an over-release inside "
        "handle_jobdispatch()/dispatch_a_job_() would have freed j while "
        "the canary was still held, making this a heap-use-after-free "
        "(this terminal site already dereferences correctly even before "
        "bug-1627's fix, which was scoped to the four retryable sites -- "
        "this check guards the same class of risk, not that specific bug)");
}

/* Task 5.3's focused proof (Req 4.2, 4.5) -- see this file's header for how
 * this differs from test_handle_jobdispatch_matches_dispatch_a_job_notify_
 * count above. A single-job batch, terminal-classified (nhosts == 0),
 * driven through the real handle_jobdispatch() entry point. */
static void test_handle_jobdispatch_terminal_no_notify_no_leak(void)
{
    xce_job_alloc_t*   allocs[1];
    xce_job_dispatch_t dispatchmsg;
    xce_request_t      req;
    job_t*             jterm;
    int                rc;

    jnd_setup();

    jterm = jnd_seed_job(JND_JOBID_TERMINAL_SOLO, "jnd-q-solo");

    /* nhosts == 0: XCE_DISPATCH_FAIL_SITE_INVALID_ALLOC, terminal --
     * mirrors this file's own JND_JOBID_TERMINAL allocation and
     * test_dispatch_fail_terminal.c's test_invalid_alloc_resolves_job_
     * to_exit, but reached this time through the real handle_jobdispatch()
     * entry point instead of dispatch_a_job_test()'s bypass. */
    allocs[0] = make_job_alloc(JND_JOBID_TERMINAL_SOLO, 0, NULL);
    TEST_ASSERT_NOT_NULL(allocs[0]);

    memset(&dispatchmsg, 0, sizeof(dispatchmsg));
    dispatchmsg.njobs  = 1;
    dispatchmsg.allocs = allocs;

    memset(&req, 0, sizeof(req));
    req.opcode           = OPCode_jobdispatch;
    req.seq              = 43;
    req.from             = &g_jnd_req_from;
    req.body_which       = Request_body_dispatchmsg;
    req.body.dispatchmsg = &dispatchmsg;

    cbrp_notify_capture_reset_ext();

    jnd_arm_ref_canary(jterm);

    /* Act: handle_jobdispatch() consumes/frees allocs[0] itself via
     * MOVE_PTR + free_JobAlloc_ptr() in its own loop -- nothing left for
     * this test to free afterward. */
    rc = handle_jobdispatch(&g_jnd, &g_jnd_req_ep, &req);
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "handle_jobdispatch's own return "
        "is unconditionally 0, unaffected by any per-job outcome");

    TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_EXIT, jterm->job.state,
        "Req 4.2: a terminally-classified dispatch failure must resolve "
        "the job to EXIT, reached this time through the real "
        "handle_jobdispatch() entry point rather than the "
        "dispatch_a_job_test() bypass test_dispatch_fail_terminal.c uses");
    TEST_ASSERT_NULL_MESSAGE(jterm->job.allocs,
        "Req 4.5: no partial allocation data may be left behind");

    /* THE assertion this task exists for: a terminal dispatch failure must
     * not ALSO produce a retry-eligible jobDispatchFail report for the same
     * failure -- mark_job_fail()'s own schedFailJob hook chain (already
     * fired above) is the sole notification. Unlike test_handle_jobdispatch_
     * matches_dispatch_a_job_notify_count's 3-job batch (where the terminal
     * job's zero contribution is inferred by the total equaling exactly the
     * retryable job's own 1 call), this single-job batch makes the zero
     * count direct: any notify_dispatch() call surfacing here can only have
     * come from this one terminal job, whether from handle_jobdispatch()'s
     * old per-job loop call (task 4.1's regression) or from dispatch_a_
     * job_() itself misclassifying/misrouting (task 3.2's regression). */
    TEST_ASSERT_EQUAL_INT_MESSAGE(0, cbrp_notify_capture_count_ext(),
        "a terminal dispatch failure must not also produce a "
        "notify_dispatch() report for this same failure (regression test "
        "for the round-2 double-notify finding, design.md's Integration "
        "Tests third bullet)");

    jnd_assert_ref_not_over_released(jterm, JND_JOBID_TERMINAL_SOLO);

    jnd_teardown();
    printf("test_handle_jobdispatch_terminal_no_notify_no_leak: PASS\n");
}

void test_jobdispatch_no_double_notify(void)
{
    UnitySetTestFile(__FILE__);
    printf("\n>> handle_jobdispatch() no-double-notify tests (Req 3.1, 4.1, "
          "4.2)\n\n");

    RUN_TEST(test_handle_jobdispatch_matches_dispatch_a_job_notify_count);
    RUN_TEST(test_handle_jobdispatch_terminal_no_notify_no_leak);
}