myxce/src/scheduler/testsuite/test_job_array_wire_e2e.c

/* test_job_array_wire_e2e.c — job-array-single-wal-record task 5.5
 *
 * Cross-daemon integration test for the Compact Scheduler/API-Server Push
 * component's scheduler-side receive path. Proves the chain everything in
 * tasks 5.1-5.4 was built to support, but that no single existing test
 * composes end to end:
 *
 *   xce_encode()  (real wire bytes, same as xcesched_push_jobarray() sends)
 *     -> xce_decode()
 *     -> extract_data()            (task 5.3b's fix)
 *     -> scheduler_req_handler()   (the REAL top-level function vtask_entry()
 *                                    registers as ctx->handler -- xcectx.c;
 *                                    never exercised by any prior job-array
 *                                    test, which all call sched_handle_dataset()
 *                                    one layer below it)
 *     -> sched_handle_dataset()
 *     -> sched_on_job_array()      (task 5.3)
 *
 * ...delivers the compact descriptor into the pend-updates queue intact and
 * unexpanded (Req 6.3, 6.4), correctly addressed by base jobid/array_size.
 *
 * REVISED (connection-loss-under-burst fix, see XCE_SCHED_OP_UPDATE_JOB_
 * ARRAY's own comment, scheduler.h): sched_on_job_array() no longer expands
 * the descriptor into N tasks, deduplicates a redelivery, or logs a
 * confirmation line itself -- all three now happen in driver.c's
 * translate_one_pending_update(), on the round-processing thread, covered by
 * test_driver_translate.c's own UPDATE_JOB_ARRAY tests
 * (test_translate_job_array_expands_into_n_events,
 * test_translate_job_array_skips_already_existing_elements,
 * test_translate_job_array_all_duplicates_returns_zero_noop). This file's
 * own two tests were revised accordingly: a redelivery of byte-identical
 * wire bytes through THIS layer alone is correctly and safely a second,
 * independent, unexpanded descriptor -- the wire/handler layer has no
 * dedup check of its own to lose; the actual dedup coverage lives in
 * test_driver_translate.c now.
 *
 * Existing coverage this test does NOT duplicate:
 *   - test_job_array_sched_push.c (task 5.1) already proves the CONTROLLER
 *     encodes exactly one compact push per multi-element notification.
 *   - test_request.c's test_extract_data_job_array_passes_through_intact
 *     (task 5.3b) already proves extract_data() itself decodes the payload
 *     correctly via a real encode/decode round trip.
 *   - test_job_array_expand.c's test_wire_dispatch_chain_delivers_job_array_
 *     descriptor (task 5.3b) already proves extract_data() ->
 *     sched_handle_dataset() compose, but stops one layer below the real
 *     registered handler.
 * This test's unique contribution: the REAL top-level dispatcher
 * (scheduler_req_handler(), not sched_handle_dataset() directly) and the
 * redelivery-under-real-wire-bytes case.
 *
 * vtask_entry() (xcectx.c) is `static` and not exported for testing -- it is
 * also the trigger normally driven by a live socket + iomgr epoll loop, which
 * this task's status report judges disproportionate to stand up fresh for a
 * single scenario (see the task 5.5 status report's TEST_STRATEGY). This
 * test instead mirrors vtask_entry()'s few lines of push-handling glue
 * verbatim (see deliver_via_real_dispatch_chain() below) -- every OTHER step
 * (encode, decode, extract_data, the handler chain itself) is the real,
 * unmodified production function, reached the same way production reaches
 * it: via xce_decode()'s output, never a hand-built dataset_t.
 *
 * Bug-catching demonstration (this task's "RED phase"): temporarily
 * reverting extract_data()'s `case XceData_data_jobArray:` (task 5.3b's fix)
 * back to falling through to `default: rc = -EINVAL` makes this test's first
 * assertion (`callnext != 0` while dataset->rows stays NULL) fail loudly --
 * confirming this test has teeth against the exact class of bug 5.3b fixed.
 */
#define UNIT_TEST 1
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <unity/unity.h>

#include "kvmgr.h"
#include "scheduler.h"

/* Pull the real handler-under-test into this TU, same white-box-but-real-
 * symbols convention as test_job_array_expand.c (handler.c is otherwise
 * self-contained; sched_init_config is its only unresolved extern). */
#include "../src/handler.c"

int sched_init_config(xce_scheduler_t* s, xce_conf4sched_t* c)
{
    (void)s;
    (void)c;
    return 0;
}

/* bug-1846: handler.c's config-push path now stashes via sched_stash_config()
 * (scheduler.c, not linked here) instead of calling sched_init_config() --
 * link-satisfying stub, same rationale as the one above. */
int sched_stash_config(xce_scheduler_t* s, xce_conf4sched_t* c)
{
    (void)s;
    (void)c;
    return 0;
}

/* sched_record_status (recording.c, scheduler-core spec, task 9.4) --
 * handler.c's OPCode_schedAdmin/recordStatus path references it; this file
 * never exercises OPCode_schedAdmin, so a link-satisfying stub is enough,
 * same rationale as sched_init_config above. */
int sched_record_status(xce_scheduler_t* sched, sched_admin_reply_t* reply)
{
    (void)sched;
    (void)reply;
    return -EINVAL;
}

/* REAL kvmgr-backed job add/get -- same rtti path production uses (mirrors
 * test_job_array_expand.c's precedent), needed so sched_on_job_array()'s
 * per-element dedup check (sched_get_job) is exercised for real. Relocated
 * to core.h (scheduler-core spec, task 4.1) -- now operate on
 * xce_sched_core_t*, not xce_scheduler_t*. */
int sched_add_job(xce_sched_core_t* core, xce_sched_job_t* j)
{
    int rc;
    lock(core->jobs, RTTI_LOCK_WRITE);
    rc = key_add(core->jobs, (void*)j->jobid, j);
    unlock(core->jobs);
    return rc;
}

xce_sched_job_t* sched_get_job(xce_sched_core_t* core, uint64_t jobid)
{
    int              rc;
    xce_sched_job_t* job = NULL;

    lock(core->jobs, RTTI_LOCK_READ);
    rc = key_find(core->jobs, (void*)jobid, (void**)&job);
    unlock(core->jobs);

    if (rc <= 0) {
        return NULL;
    }

    return job;
}

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

static void fixture_init(xce_scheduler_t* sched)
{
    kvmgr_params_t jobs_params = { .key_type   = KVMGR_KEY_INT,
                                   .nslots     = 59,
                                   .delta      = 16,
                                   .datakiller = NULL };

    memset(sched, 0, sizeof(*sched));
    mylock_init(&(sched->core.lock));
    sched->core.pend_updates[0] = (dataq_t*)calloc(1, sizeof(dataq_t));
    dataq_init(sched->core.pend_updates[0], NULL);
    sched->core.pend_updates[1] = (dataq_t*)calloc(1, sizeof(dataq_t));
    dataq_init(sched->core.pend_updates[1], NULL);
    sched->core.jobs = makeInstance(&xkvmgr_type, &jobs_params);
}

/* Pop one pending update request, asserting it is an UPDATE_JOB_ARRAY op;
 * hands back the raw job_array descriptor (caller now owns it -- free via
 * free_JobSubmit_ptr). Returns NULL if the queue is empty. Mirrors
 * test_job_array_expand.c's identical helper. */
static xce_job_submit_t* pop_job_array_update(xce_scheduler_t* sched)
{
    data_item_t* it = dataq_deq(sched->core.pend_updates[0], NULL);
    xce_sched_update_request_t* req;
    xce_job_submit_t*           job_array;

    if (it == NULL) {
        return NULL;
    }

    req = ITPTR_TO_UPDATE_REQ(it);
    TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_UPDATE_JOB_ARRAY, req->op);

    job_array = req->u.jobarrayupdate.job_array;
    free(req);
    return job_array;
}

/* ---------------------------------------------------------------------- */
/* Real wire construction: the exact message xcesched_push_jobarray()      */
/* (sched.c) sends -- OPCode_push / Request_body_pushmsg carrying one      */
/* Data.jobArray descriptor.                                                */
/* ---------------------------------------------------------------------- */

static void build_encoded_push(io_packet_t** out_pkt, uint64_t base_jobid,
                               uint64_t array_size,
                               job_submit_data_t* submit_storage)
{
    xce_job_submit_t job_array;
    xce_data_t       data;
    xce_request_t    request;
    io_packet_t*     pkt;

    memset(submit_storage, 0, sizeof(*submit_storage));
    submit_storage->queue    = "normal";
    submit_storage->user     = "tester";
    submit_storage->mintasks = 1;
    submit_storage->maxtasks = 1;
    submit_storage->resreq   = "type == local";

    memset(&job_array, 0, sizeof(job_array));
    job_array.jobid      = base_jobid;
    job_array.array_size = array_size;
    job_array.submit     = submit_storage;

    memset(&data, 0, sizeof(data));
    data.table          = XceData_data_jobArray;
    data.data.job_array = &job_array;

    memset(&request, 0, sizeof(request));
    request.opcode       = OPCode_push;
    request.body_which   = Request_body_pushmsg;
    request.body.pushmsg = &data;

    pkt = (io_packet_t*)calloc(1, sizeof(*pkt));
    TEST_ASSERT_NOT_NULL(pkt);
    TEST_ASSERT_EQUAL_INT(1, xce_encode(&request, pkt));

    *out_pkt = pkt;
}

/* Mirrors vtask_entry()'s (xcectx.c, static) push-handling glue exactly:
 *   1. xce_decode() the packet -- fresh, independent decode every call, so
 *      a "second delivery" is genuinely re-derived from wire bytes, not a
 *      reused in-memory struct.
 *   2. Allocate a dataset_t, call the REAL extract_data() (task 5.3b).
 *   3. On success (vtask_entry(): `if (callnext)`, true for ANY nonzero
 *      return, negative included -- this is the exact amplifier that let
 *      5.3b's bug through silently), free the now-hollow Data container via
 *      the same FREEUP() production macro vtask_entry() uses, then install
 *      the reassembled dataset into body.datamsg/body_which exactly as
 *      vtask_entry()'s MOVE_PTR does.
 *   4. Call the REAL top-level scheduler_req_handler() -- never reached by
 *      any prior job-array test.
 * Returns scheduler_req_handler()'s rc; *out_ds receives the dataset (caller
 * owns dataset->rows cleanup via free_JobSubmit_ptr, matching
 * sched_handle_dataset()'s jobArray case, which deliberately never nulls
 * dataset->rows -- see that function's own comment). */
static int deliver_via_real_dispatch_chain(io_packet_t* pkt,
                                           xce_scheduler_t* sched,
                                           dataset_t** out_ds)
{
    xce_request_t decoded;
    dataset_t*    ds;
    int           callnext;
    int           rc;

    memset(&decoded, 0, sizeof(decoded));
    TEST_ASSERT_EQUAL_INT(0, xce_decode(pkt, &decoded));
    TEST_ASSERT_EQUAL_INT(Request_body_pushmsg, decoded.body_which);
    TEST_ASSERT_NOT_NULL(decoded.body.pushmsg);

    ds = (dataset_t*)calloc(1, sizeof(*ds));
    TEST_ASSERT_NOT_NULL(ds);
    ds->table = decoded.body.pushmsg->table;

    callnext = extract_data(ds, decoded.body.pushmsg);
    TEST_ASSERT_TRUE_MESSAGE(
        callnext != 0,
        "extract_data() must return nonzero (vtask_entry()'s `if (callnext)` "
        "treats ANY nonzero -- including a negative errno -- as reassembly "
        "complete); a zero return here would mean this test's own harness is "
        "broken, not exercising the bug class");

    FREEUP(decoded.body.pushmsg);
    decoded.body_which   = Request_body_datamsg;
    decoded.body.datamsg = ds;

    rc = scheduler_req_handler(&decoded, sched);

    decoded.body.datamsg = NULL; /* caller owns *ds; avoid double free below */
    xce_free_request(&decoded);

    *out_ds = ds;
    return rc;
}

/* ── Test 1: one real wire push -> the compact descriptor is queued intact */
/* and unexpanded (Req 6.3, 6.4) -- expansion itself is driver.c's concern  */
/* now, covered by test_driver_translate.c.                                */

static void test_one_push_enqueues_job_array_descriptor(void)
{
    xce_scheduler_t   sched;
    job_submit_data_t submit_storage;
    io_packet_t*      pkt = NULL;
    dataset_t*        ds  = NULL;
    xce_job_submit_t* popped;

    fixture_init(&sched);

    build_encoded_push(&pkt, 42000, 6, &submit_storage);

    TEST_ASSERT_TRUE(
        deliver_via_real_dispatch_chain(pkt, &sched, &ds) >= 0);

    popped = pop_job_array_update(&sched);
    TEST_ASSERT_NOT_NULL(popped);
    TEST_ASSERT_EQUAL_UINT64(42000, popped->jobid);
    TEST_ASSERT_EQUAL_UINT64(6, popped->array_size);
    TEST_ASSERT_EQUAL_STRING("normal", popped->submit->queue);
    free_JobSubmit_ptr(&popped);

    free_JobSubmit_ptr((xce_job_submit_t**)&ds->rows);
    free(ds);
    printf("test_one_push_enqueues_job_array_descriptor: PASS\n");
}

/* ── Test 2: redelivery of the IDENTICAL encoded bytes (independently      */
/* re-decoded, simulating a reconnect replay) -- REVISED (connection-loss-  */
/* under-burst fix): the wire/handler layer no longer deduplicates at all   */
/* (see XCE_SCHED_OP_UPDATE_JOB_ARRAY's own comment, scheduler.h), so a     */
/* redelivery through THIS layer alone correctly and safely produces a      */
/* second, independent, unexpanded descriptor -- dedup coverage now lives   */
/* in test_driver_translate.c's own                                        */
/* test_translate_job_array_skips_already_existing_elements /               */
/* test_translate_job_array_all_duplicates_returns_zero_noop (Req 6.5).     */

static void test_redelivery_enqueues_independent_second_descriptor(void)
{
    xce_scheduler_t   sched;
    job_submit_data_t submit_storage;
    io_packet_t*      pkt1 = NULL;
    io_packet_t*      pkt2 = NULL;
    dataset_t*        ds1  = NULL;
    dataset_t*        ds2  = NULL;
    xce_job_submit_t* popped1;
    xce_job_submit_t* popped2;

    fixture_init(&sched);

    /* Build the wire bytes twice (two independent xce_encode() calls with
     * identical inputs) so each delivery genuinely goes through its own
     * encode and decode -- not one decode fed to the handler twice. */
    build_encoded_push(&pkt1, 43000, 4, &submit_storage);
    build_encoded_push(&pkt2, 43000, 4, &submit_storage);

    /* First delivery. */
    TEST_ASSERT_TRUE(deliver_via_real_dispatch_chain(pkt1, &sched, &ds1) >= 0);
    popped1 = pop_job_array_update(&sched);
    TEST_ASSERT_NOT_NULL(popped1);
    TEST_ASSERT_EQUAL_UINT64(43000, popped1->jobid);
    TEST_ASSERT_EQUAL_UINT64(4, popped1->array_size);
    free_JobSubmit_ptr(&popped1);
    free_JobSubmit_ptr((xce_job_submit_t**)&ds1->rows);
    free(ds1);

    /* Second, byte-identical delivery (simulated redelivery/reconnect): a
     * second, independent descriptor -- correct and safe by design (the
     * downstream dedup check, now in driver.c, runs against core->jobs,
     * which this white-box TU never populates from a queued-but-not-yet-
     * translated request). */
    TEST_ASSERT_TRUE(deliver_via_real_dispatch_chain(pkt2, &sched, &ds2) >= 0);
    popped2 = pop_job_array_update(&sched);
    TEST_ASSERT_NOT_NULL(popped2);
    TEST_ASSERT_EQUAL_UINT64(43000, popped2->jobid);
    TEST_ASSERT_EQUAL_UINT64(4, popped2->array_size);
    free_JobSubmit_ptr(&popped2);

    free_JobSubmit_ptr((xce_job_submit_t**)&ds2->rows);
    free(ds2);
    printf("test_redelivery_enqueues_independent_second_descriptor: PASS\n");
}

int main(int argc, char* argv[])
{
    (void)argc;
    (void)argv;
    UNITY_BEGIN();
    RUN_TEST(test_one_push_enqueues_job_array_descriptor, 0);
    RUN_TEST(test_redelivery_enqueues_independent_second_descriptor, 0);
    return UNITY_END();
}