myxce/src/controller/testsuite/test_job_array_sched_push.c

/* test_job_array_sched_push.c — job-array-single-wal-record task 5.1
 *
 * Unit tests for the Compact Scheduler/API-Server Push component's
 * controller-side scheduler push (design.md "Compact Scheduler/API-Server
 * Push"; Req 6.1, 6.3, 6.4):
 *
 *   - A synthetic multi-element HOOK_DETAIL_JOB_NEW fire (the exact shape
 *     apply_committed_entry()'s WAL_JOB_ARRAY_SUBMIT case produces, task 3.2 —
 *     one AFTER_EVENT fire, one xlist_t carrying every constructed element)
 *     produces EXACTLY ONE queued push packet, encoding the compact
 *     Data.jobArray descriptor (base jobid + array size + one shared spec) —
 *     never N individually-encoded Job/SchedJob structs.
 *   - A synthetic single-element fire (ordinary single-job submission) is
 *     byte-for-byte unaffected: still the existing Data.schedjobs push via
 *     xcesched_push_joblist(), unchanged (Req 8.1).
 *
 * Fixture: sched_job_new_fun()/xcesched_push_joblist()/xcesched_push_jobarray()
 * are static to sched.c, so this file drives them through the same public
 * entry points production code uses: on_scheduler_join() installs the hook
 * (its `hook_installed` static is once-per-process, so ALL tests in this
 * file share ONE cluster_t/hooksys wired exactly once — see jsp_setup) and
 * xce_run_hooks() fires it exactly as apply_committed_entry() does. The
 * outbound push itself is observed the same way test_commit_barrier_rollout.c
 * observes queued replies (cb5_scan_replies): a zeroed, fd=-1 stack
 * endpoint_t queues packets on ep->outq without touching a real socket
 * (endpoint_update_epoll no-ops, zeroed pthread mutexes are valid); this
 * test drains and raw-capnp-decodes those packets instead.
 *
 * Job list elements are plain job_t structs with only the two fields either
 * push function reads (.job.jobid, .job.submit) plus submit_lsn = 0 (makes
 * job_submit_committed() unconditionally true, matching a freshly-committed
 * array's elements without needing a real persist_t) — NOT routed through
 * create_job()/jobj.c, whose N-way construction is already covered by
 * test_job_array_apply.c.
 */

#include <capnp_c.h>
#include <errno.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>

#include <unity/unity.h>

#include "cluster.h"
#include "controller.capnp.h"
#include "hook.h"
#include "netio.h"
#include "request.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xlist.h"

extern int on_scheduler_join(cluster_t*);

/* This file builds a STANDALONE test binary (meson.build's sched_push_unit_
 * test) linking the real sched.c + its real dataex.c/egress.c dependencies —
 * see meson.build's comment for why this can't be folded into the shared
 * controller_unit_test binary. opget.c's check_worker_() is sched.c's only
 * remaining unresolved reference (from xcesched_push_workers(), reached only
 * through on_scheduler_join()'s push-on-join branch, which this file's setup
 * deliberately never triggers at runtime — c->sched.ep is NULL during that
 * call — but the symbol must still resolve at link time). Stubbed locally
 * rather than linking all of opget.c, mirroring opget.c's own trivial body. */
int check_worker_(xce_worker_t* w) { return (w != NULL) ? 1 : 0; }

/* bug-1693: sched.c's handle_scheduler_leave() now calls
 * schedadmin_signal_disconnect() (defined in cluster.c, which this
 * standalone binary does not link -- see this file's own comment above for
 * why). This test never triggers a scheduler-leave/disconnect path, so the
 * real behavior is never exercised here; the symbol only needs to resolve
 * at link time, mirroring check_worker_()'s identical stub rationale just
 * above. */
int schedadmin_signal_disconnect(void) { return 0; }

/* ------------------------------------------------------------------ */
/* Fixture                                                              */
/* ------------------------------------------------------------------ */

static cluster_t  g_jsp_c;
static endpoint_t g_jsp_ep;
static int        g_jsp_hook_wired;

static void jsp_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 jsp_setup(void)
{
    if (!g_jsp_hook_wired) {
        memset(&g_jsp_c, 0, sizeof(g_jsp_c));
        TEST_ASSERT_EQUAL_INT(
            0, xce_hook_init(&g_jsp_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
        /* sched.ep == NULL here so on_scheduler_join's push-on-join branch
         * (xcesched_push_configs/workers/jobs) is a no-op — this call's only
         * effect is installing sched_job_hook_fun (and siblings) into
         * g_jsp_c.hooksys, once, process-wide (hook_installed is a static
         * inside on_scheduler_join). */
        g_jsp_c.sched.ep = NULL;
        TEST_ASSERT_EQUAL_INT(0, on_scheduler_join(&g_jsp_c));
        g_jsp_hook_wired = 1;
    }

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

    g_jsp_c.sched.ep         = &g_jsp_ep;
    g_jsp_c.running          = 1;
    g_jsp_c.myhost           = &g_jsp_c.myhost_info;
    g_jsp_c.myhost_info.name = (char*)"ctrl-jsp";
    g_jsp_c.myhost_info.key  = (char*)"";
}

static void jsp_teardown(void)
{
    jsp_drain_ep(&g_jsp_ep);
    g_jsp_c.sched.ep = NULL;
}

/* ------------------------------------------------------------------ */
/* Helpers                                                              */
/* ------------------------------------------------------------------ */

static job_submit_data_t g_jsp_submit_data;

static void jsp_make_job(job_t* out, uint64_t jobid)
{
    memset(out, 0, sizeof(*out));
    out->job.jobid  = jobid;
    out->job.submit = &g_jsp_submit_data;
    out->submit_lsn
        = 0; /* job_submit_committed(): submit_lsn==0 => always exportable */
}

static int jsp_fire_job_new(xlist_t* l)
{
    return xce_run_hooks(&g_jsp_c.hooksys, XCE_CONTROLLER_HOOK_JOB,
                         HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_NEW, &g_jsp_c, l);
}

typedef struct {
    int      n_pushes;
    int      last_data_which;
    uint64_t last_jobid;
    uint64_t last_array_size;
} jsp_push_obs_t;

/* Raw-capnp-parse every queued push packet on ep->outq (mirrors
 * test_commit_barrier_rollout.c's cb5_scan_replies). Drains the queue. */
static void jsp_scan_pushes(endpoint_t* ep, jsp_push_obs_t* obs)
{
    data_item_t* it;

    memset(obs, 0, sizeof(*obs));
    while ((it = endpoint_deq(ep)) != NULL) {
        io_packet_t*   pkt = ITPTR_TO_PACKET(it);
        struct capn    c;
        struct Request rq;
        Request_ptr    rp;

        capn_init_mem(&c, (const uint8_t*)pkt->data, pkt->sz, 0);
        rp.p = capn_getp(capn_root(&c), 0, 1);
        read_Request(&rq, rp);
        if (rq.body_which == Request_body_pushmsg) {
            struct XceData d;

            read_XceData(&d, rq.body.pushmsg);
            obs->n_pushes += 1;
            obs->last_data_which = (int)d.data_which;
            if (d.data_which == XceData_data_jobArray) {
                struct JobSubmit js;

                read_JobSubmit(&js, d.data.jobArray);
                obs->last_jobid      = js.jobid;
                obs->last_array_size = js.arraySize;
            }
        }
        capn_free(&c);
        free_packet(pkt);
        free(pkt);
    }
}

/* ------------------------------------------------------------------ */
/* Test 1: multi-element fire -> exactly one compact Data.jobArray push */
/* ------------------------------------------------------------------ */

static void test_multi_element_job_new_produces_one_compact_jobarray_push(void)
{
    job_t          jobs[5];
    void*          storage[5];
    xlist_t        l = { .count = 5, .caps = 5, .delta = 5, .data = storage };
    jsp_push_obs_t obs;
    int            rc;

    jsp_setup();

    for (int i = 0; i < 5; i++) {
        jsp_make_job(&jobs[i], 5000 + (uint64_t)i);
        storage[i] = &jobs[i].job;
    }

    rc = jsp_fire_job_new(&l);
    TEST_ASSERT_EQUAL_INT(0, rc);

    jsp_scan_pushes(&g_jsp_ep, &obs);

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        1, obs.n_pushes,
        "a 5-element array notification must produce exactly ONE push packet, "
        "not one per element (Req 6.1, 6.4)");
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        XceData_data_jobArray, obs.last_data_which,
        "the push must encode the compact Data.jobArray descriptor, not "
        "Data.schedjobs/Data.jobs (N individually-encoded job structs)");
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(
        5000, obs.last_jobid,
        "base jobid must be the array's first (lowest) jobid");
    TEST_ASSERT_EQUAL_UINT64_MESSAGE(
        5, obs.last_array_size,
        "arraySize must be the notification's element count");

    jsp_teardown();
    printf("test_multi_element_job_new_produces_one_compact_jobarray_push: "
           "PASS\n");
}

/* ------------------------------------------------------------------ */
/* Test 2: single-element fire -> unmodified per-job Data.schedjobs push */
/* ------------------------------------------------------------------ */

static void test_single_element_job_new_is_unaffected(void)
{
    job_t          job;
    void*          storage[1];
    xlist_t        l = { .count = 1, .caps = 1, .delta = 1, .data = storage };
    jsp_push_obs_t obs;
    int            rc;

    jsp_setup();

    jsp_make_job(&job, 9001);
    storage[0] = &job.job;

    rc = jsp_fire_job_new(&l);
    TEST_ASSERT_EQUAL_INT(0, rc);

    jsp_scan_pushes(&g_jsp_ep, &obs);

    TEST_ASSERT_EQUAL_INT_MESSAGE(
        1, obs.n_pushes,
        "an ordinary single-job notification must still produce exactly one "
        "push packet, unchanged");
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        XceData_data_schedjobs, obs.last_data_which,
        "a single-element notification must go through the EXISTING, "
        "UNMODIFIED xcesched_push_joblist() path (Data.schedjobs), never the "
        "new compact jobArray path (Req 8.1)");

    jsp_teardown();
    printf("test_single_element_job_new_is_unaffected: PASS\n");
}

/* ------------------------------------------------------------------ */
/* Standalone Unity main (this binary is its own test executable — see  */
/* meson.build's sched_push_unit_test)                                  */
/* ------------------------------------------------------------------ */

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

int main(void)
{
    printf("\n>> job-array-single-wal-record task 5.1: compact scheduler "
           "push\n\n");

    UNITY_BEGIN();

    RUN_TEST(test_multi_element_job_new_produces_one_compact_jobarray_push);
    RUN_TEST(test_single_element_job_new_is_unaffected);

    return UNITY_END();
}