myxce/src/scheduler/testsuite/test_job_array_expand.c

/*
 * test_job_array_expand.c - white-box tests for the scheduler's compact
 * job-array push receive path (job-array-single-wal-record task 5.3, Req
 * 6.3, 6.4, 6.5, 8.4).
 *
 * 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 compact descriptor into N tasks itself -- it just takes ownership and
 * enqueues it as a single XCE_SCHED_OP_UPDATE_JOB_ARRAY request. The actual
 * per-element expansion (and its dedup check) now runs in driver.c's
 * translate_one_pending_update(), covered by test_driver_translate.c's own
 * UPDATE_JOB_ARRAY tests. This file now covers only: (a) sched_on_job_array()
 * /sched_handle_dataset()'s own input validation and ownership-transfer
 * bookkeeping, and (b) that the wire-decode -> handler -> queue chain
 * delivers the descriptor's fields correctly, unmodified.
 *
 * Strategy: pull the real (static) sched_handle_dataset / sched_on_job_array
 * into this TU via #include "../src/handler.c" (same white-box pattern as
 * test_handlers.c). sched_add_job/sched_get_job stubs below are unused by
 * anything this file still exercises (kept, harmless, for parity with other
 * handler.c white-box test files that DO need them).
 */
#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. handler.c is otherwise
 * self-contained for the dataset dispatch path (enq_pend_update is static
 * within it); the only scheduler-internal extern it references that lives
 * elsewhere is sched_init_config, stubbed below (same as test_handlers.c). */
#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_update.c's precedent for exercising sched_get_job "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)
{
    printf("\n-----------------------------------------------------\n");
    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);
}

/* Builds a heap-allocated, heap-owned-strings (jobArray, submit) pair -- the
 * same shape decode_JobSubmit_ptr() produces on the real wire-decode path
 * (controller.capnp.c), not a stack/string-literal stand-in. Task 7.3 made
 * sched_handle_dataset()'s XceData_data_jobArray case free_JobSubmit_ptr() its
 * dataset->rows unconditionally, so any fixture routed through that handler
 * must be genuinely heap-owned or the free segfaults on stack memory /
 * string literals. Caller owns the returned pointer until it hands it to
 * sched_handle_dataset() (which frees it) or frees it directly itself. */
static xce_job_submit_t* init_job_array(uint64_t base_jobid,
                                        uint64_t array_size)
{
    job_submit_data_t* submit = (job_submit_data_t*)calloc(1, sizeof(*submit));
    xce_job_submit_t*  ja     = (xce_job_submit_t*)calloc(1, sizeof(*ja));

    submit->queue    = strdup("normal");
    submit->user     = strdup("tester");
    submit->mintasks = 1;
    submit->maxtasks = 1;
    submit->resreq   = strdup("type == local");

    ja->jobid      = base_jobid;
    ja->array_size = array_size;
    ja->submit     = submit;

    return ja;
}

/* 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 to mirror release_translated_payload()'s own cleanup on
 * the failure/never-translated path). Returns NULL if the queue is empty. */
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;
    TEST_ASSERT_EQUAL_INT(0, req->u.jobarrayupdate.expanded_njobs);
    TEST_ASSERT_NULL(req->u.jobarrayupdate.expanded_jobs);
    free(req);
    return job_array;
}

/* ── Connection-loss-under-burst fix: a compact Data.jobArray push is now
 * queued UNEXPANDED -- ownership of the raw descriptor transfers to the
 * XCE_SCHED_OP_UPDATE_JOB_ARRAY request as-is, no per-element work happens
 * on this (request-handling) thread. Expansion into N tasks is
 * test_driver_translate.c's concern now (translate_one_pending_update()). ── */

static void test_job_array_push_enqueues_raw_descriptor_unexpanded(void)
{
    xce_scheduler_t   sched;
    xce_job_submit_t* ja;
    dataset_t         dataset;
    xce_job_submit_t* popped;

    fixture_init(&sched);
    ja = init_job_array(5000, 3);

    memset(&dataset, 0, sizeof(dataset));
    dataset.table = XceData_data_jobArray;
    dataset.rows  = (void*)ja;

    TEST_ASSERT_TRUE(sched_handle_dataset(&sched, &dataset) >= 0);

    /* Ownership transferred to the queued request -- sched_handle_dataset()
     * must not have freed it and must have nulled dataset.rows out. */
    TEST_ASSERT_NULL(dataset.rows);

    popped = pop_job_array_update(&sched);
    TEST_ASSERT_EQUAL_PTR(ja, popped); /* same descriptor, not a clone */
    TEST_ASSERT_EQUAL_UINT64(5000, popped->jobid);
    TEST_ASSERT_EQUAL_UINT64(3, popped->array_size);
    TEST_ASSERT_EQUAL_STRING("normal", popped->submit->queue);

    free_JobSubmit_ptr(&popped);
}

/* ── Guard-clause coverage on the decode step itself ── */

static void test_job_array_null_submit_returns_einval(void)
{
    xce_scheduler_t  sched;
    xce_job_submit_t ja;

    fixture_init(&sched);

    memset(&ja, 0, sizeof(ja));
    ja.jobid      = 1;
    ja.array_size = 3;
    ja.submit     = NULL;

    TEST_ASSERT_EQUAL_INT(-EINVAL, sched_on_job_array(&sched, &ja));
}

static void test_job_array_zero_size_returns_einval(void)
{
    xce_scheduler_t   sched;
    xce_job_submit_t* ja;

    fixture_init(&sched);
    ja = init_job_array(42, 0);

    TEST_ASSERT_EQUAL_INT(-EINVAL, sched_on_job_array(&sched, ja));

    /* unlike sched_handle_dataset(), sched_on_job_array() never takes
     * ownership of job_array -- this test built it directly, so it frees it
     * directly. */
    free_JobSubmit_ptr(&ja);
}

static void test_job_array_null_descriptor_returns_einval(void)
{
    xce_scheduler_t sched;
    fixture_init(&sched);

    TEST_ASSERT_EQUAL_INT(-EINVAL, sched_on_job_array(&sched, NULL));
}

/* ── job-array-single-wal-record task 5.3b: full wire-dispatch chain
 * (REVISED for the connection-loss-under-burst fix) ──────────────────────
 * Every test above feeds a stack-built xce_job_submit_t straight into
 * sched_handle_dataset(), bypassing the transport layer entirely -- the same
 * white-box convention every sibling handler.c test uses (task 5.3's own
 * Implementation Notes call this out explicitly: those tests were never
 * intended to prove real-wire delivery). This test drives the REAL
 * xce_encode()/xce_decode()/extract_data() path (src/lib/request.c, task
 * 5.3b's fix) with the exact wire shape xcesched_push_jobarray() (sched.c)
 * sends -- OPCode_push / Request_body_pushmsg carrying one Data.jobArray
 * descriptor -- then hands extract_data()'s reassembled dataset_t to
 * sched_handle_dataset(), proving the sender-payload -> wire-dispatch ->
 * receiver-enqueue chain delivers the descriptor's fields correctly, intact
 * end to end -- expansion itself is no longer this chain's job (driver.c's
 * concern now), so this only proves the queued descriptor. */

static void test_wire_dispatch_chain_delivers_job_array_descriptor(void)
{
    xce_scheduler_t   sched;
    job_submit_data_t submit;
    xce_job_submit_t  job_array;
    xce_data_t        data;
    xce_request_t     request;
    xce_request_t     decoded;
    io_packet_t*      pkt;
    dataset_t         dataset;
    xce_job_submit_t* popped;

    fixture_init(&sched);

    /* Arrange: the exact wire message xcesched_push_jobarray() sends. */
    memset(&submit, 0, sizeof(submit));
    submit.queue    = "normal";
    submit.user     = "tester";
    submit.mintasks = 1;
    submit.maxtasks = 1;
    submit.resreq   = "type == local";

    memset(&job_array, 0, sizeof(job_array));
    job_array.jobid      = 8000;
    job_array.array_size = 4;
    job_array.submit     = &submit;

    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));

    memset(&decoded, 0, sizeof(decoded));
    TEST_ASSERT_EQUAL_INT(0,
                          xce_decode(pkt, &decoded)); /* consumes/frees pkt */

    /* Act 1: the REAL transport dispatcher (task 5.3b's fix) -- exactly what
     * vtask_entry() (xcectx.c) calls on every inbound push. */
    memset(&dataset, 0, sizeof(dataset));
    dataset.table = decoded.body.pushmsg->table;
    TEST_ASSERT_EQUAL_INT(1, extract_data(&dataset, decoded.body.pushmsg));
    TEST_ASSERT_NOT_NULL(dataset.rows);

    /* Act 2: hand the reassembled dataset straight to the receiver's own
     * handler (task 5.3, unmodified) -- proving the two layers compose. */
    TEST_ASSERT_TRUE(sched_handle_dataset(&sched, &dataset) >= 0);

    /* Assert: the decoded descriptor was queued unexpanded, fields intact,
     * correctly addressed via base_jobid + array_size. */
    popped = pop_job_array_update(&sched);
    TEST_ASSERT_NOT_NULL(popped);
    TEST_ASSERT_EQUAL_UINT64(8000, popped->jobid);
    TEST_ASSERT_EQUAL_UINT64(4, popped->array_size);
    TEST_ASSERT_EQUAL_STRING("normal", popped->submit->queue);
    free_JobSubmit_ptr(&popped);

    /* Cleanup: extract_data() transferred ownership of the decoded
     * xce_job_submit_t into dataset.rows (nulling the source pointer inside
     * decoded.body.pushmsg); sched_handle_dataset()'s jobArray case
     * transfers that ownership again, into the queued request, and nulls
     * dataset.rows out -- nothing left for this test to free there.
     * decoded.body.pushmsg is a hollow container, safe to free via the
     * normal request teardown. */
    TEST_ASSERT_NULL(dataset.rows);
    xce_free_request(&decoded);
}

int main(int argc, char* argv[])
{
    (void)argc;
    (void)argv;
    UNITY_BEGIN();
    RUN_TEST(test_job_array_push_enqueues_raw_descriptor_unexpanded, 0);
    RUN_TEST(test_job_array_null_submit_returns_einval, 0);
    RUN_TEST(test_job_array_zero_size_returns_einval, 0);
    RUN_TEST(test_job_array_null_descriptor_returns_einval, 0);
    RUN_TEST(test_wire_dispatch_chain_delivers_job_array_descriptor, 0);
    return UNITY_END();
}