/* 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 exactly N correctly-addressed scheduler tasks and fires task * 5.3's confirmation log line exactly once (Req 6.3, 6.4, 8.4); and that an * identical byte-for-byte SECOND delivery (independently re-decoded from the * same encoded bytes -- the wire-level equivalent of "the same push arrives * twice", e.g. after a reconnect) creates zero new tasks and logs zero * additional confirmation lines (Req 6.5, 8.4). * * 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_n_tasks * (task 5.3b) already proves extract_data() -> sched_handle_dataset() * compose, but stops one layer below the real registered handler and * never exercises redelivery through the wire path or the log line. * This test's unique contribution: the REAL top-level dispatcher * (scheduler_req_handler(), not sched_handle_dataset() directly), the * redelivery-under-real-wire-bytes case, and the confirmation log line's * actual emission count. * * 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, * and the subsequent job-count assertions fail too (zero tasks created) -- * confirming this test has teeth against the exact class of bug 5.3b fixed. */ #define UNIT_TEST 1 #include #include #include #include #include #include #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; } /* 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 op; hands * back njobs/jobs (caller now owns them -- free via free_Job_ptr + free() to * mirror sched_pend_update_killer). Returns -1 if the queue is empty. Mirrors * test_job_array_expand.c's identical helper. */ static int pop_job_update(xce_scheduler_t* sched, xce_job_t*** jobs_out) { data_item_t* it = dataq_deq(sched->core.pend_updates[0], NULL); xce_sched_update_request_t* req; int njobs; if (it == NULL) { return -1; } req = ITPTR_TO_UPDATE_REQ(it); TEST_ASSERT_EQUAL_INT(XCE_SCHED_OP_UPDATE_JOB, req->op); njobs = req->u.jobupdate.njobs; if (jobs_out != NULL) { *jobs_out = req->u.jobupdate.jobs; } free(req); return njobs; } /* Materializes a popped job array into sched->jobs (mirroring what the * async consumer, scheduler.c's sched_update_jobs -> sched_add_job, not * linked into this white-box TU, would do) -- so a SECOND, redelivered * push's dedup check sees the first delivery's elements as already * existing. Frees each popped xce_job_t and the array itself, same * simulated-reconnect technique test_job_array_expand.c's * test_job_array_redelivery_does_not_duplicate uses (there via * seed_existing_job, here folded into one step since njobs is already * known). */ static void materialize_popped_jobs(xce_scheduler_t* sched, xce_job_t** jobs, int njobs) { int i; for (i = 0; i < njobs; i++) { xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job)); job->jobid = jobs[i]->jobid; sched_add_job(&(sched->core), job); free_Job_ptr(&jobs[i]); } free(jobs); } /* ---------------------------------------------------------------------- */ /* Early-print log capture: sched_on_job_array()'s confirmation line uses */ /* INFO() (tlog_info -> TLOG_INFO). No test binary in this codebase calls */ /* tlog_init(), so every INFO() call routes through tlog's pre-init */ /* "early print" fallback (tlog.c's _tlog_early_print(), confirmed via the */ /* vendored tlog source). tlog_reg_early_printf_callback() lets a test */ /* intercept that fallback synchronously, on the calling thread -- no */ /* tlog_init()/tlog_exit() lifecycle, so it cannot disturb any other test */ /* sharing this process. va_copy() is used so the caller's own va_list is */ /* never consumed, in case tlog's own early-print path also reads it. */ /* ---------------------------------------------------------------------- */ static int g_confirm_log_count = 0; static void wire_e2e_early_print_cb(struct tlog_loginfo* loginfo, const char* format, va_list ap) { char buf[512]; va_list ap_copy; if (loginfo == NULL || loginfo->level != TLOG_INFO) { return; } va_copy(ap_copy, ap); vsnprintf(buf, sizeof(buf), format, ap_copy); va_end(ap_copy); if (strstr(buf, "job array push received and expanded") != NULL) { g_confirm_log_count++; } } static void start_log_capture(void) { g_confirm_log_count = 0; tlog_set_early_printf(1, 1, 0); tlog_reg_early_printf_callback(wire_e2e_early_print_cb); } static void stop_log_capture(void) { tlog_reg_early_printf_callback(NULL); } /* ---------------------------------------------------------------------- */ /* 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 -> N correctly-addressed tasks, log fires */ /* exactly once (Req 6.3, 6.4, 8.4) */ static void test_one_push_produces_n_tasks_and_logs_once(void) { xce_scheduler_t sched; job_submit_data_t submit_storage; io_packet_t* pkt = NULL; dataset_t* ds = NULL; xce_job_t** jobs = NULL; int njobs; int i; fixture_init(&sched); start_log_capture(); build_encoded_push(&pkt, 42000, 6, &submit_storage); TEST_ASSERT_TRUE( deliver_via_real_dispatch_chain(pkt, &sched, &ds) >= 0); njobs = pop_job_update(&sched, &jobs); TEST_ASSERT_EQUAL_INT(6, njobs); for (i = 0; i < njobs; i++) { TEST_ASSERT_EQUAL_UINT64(42000 + (uint64_t)i, jobs[i]->jobid); TEST_ASSERT_EQUAL_STRING("normal", jobs[i]->submit->queue); free_Job_ptr(&jobs[i]); } free(jobs); TEST_ASSERT_EQUAL_INT_MESSAGE( 1, g_confirm_log_count, "task 5.3's confirmation log line must fire exactly once for one " "successful array expansion"); stop_log_capture(); free_JobSubmit_ptr((xce_job_submit_t**)&ds->rows); free(ds); printf("test_one_push_produces_n_tasks_and_logs_once: PASS\n"); } /* ── Test 2: redelivery of the IDENTICAL encoded bytes (independently */ /* re-decoded, simulating a reconnect replay) creates zero new tasks and */ /* zero additional log lines (Req 6.5, 8.4) */ static void test_redelivery_creates_no_new_tasks_and_logs_nothing(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_t** jobs = NULL; int njobs; 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); start_log_capture(); /* First delivery: 4 new tasks. */ TEST_ASSERT_TRUE(deliver_via_real_dispatch_chain(pkt1, &sched, &ds1) >= 0); njobs = pop_job_update(&sched, &jobs); TEST_ASSERT_EQUAL_INT(4, njobs); materialize_popped_jobs(&sched, jobs, njobs); TEST_ASSERT_EQUAL_INT(1, g_confirm_log_count); free_JobSubmit_ptr((xce_job_submit_t**)&ds1->rows); free(ds1); /* Second, byte-identical delivery (simulated redelivery/reconnect): * every index (43000..43003) already exists in sched->jobs -- must be a * pure no-op, and must NOT log a second confirmation line. */ TEST_ASSERT_EQUAL_INT( 0, deliver_via_real_dispatch_chain(pkt2, &sched, &ds2)); TEST_ASSERT_EQUAL_INT_MESSAGE( -1, pop_job_update(&sched, &jobs), "redelivery must enqueue nothing new"); TEST_ASSERT_EQUAL_INT_MESSAGE( 1, g_confirm_log_count, "redelivery must not add a second confirmation log line -- still " "exactly 1 total, from the first delivery only"); stop_log_capture(); free_JobSubmit_ptr((xce_job_submit_t**)&ds2->rows); free(ds2); printf("test_redelivery_creates_no_new_tasks_and_logs_nothing: PASS\n"); } int main(int argc, char* argv[]) { (void)argc; (void)argv; UNITY_BEGIN(); RUN_TEST(test_one_push_produces_n_tasks_and_logs_once, 0); RUN_TEST(test_redelivery_creates_no_new_tasks_and_logs_nothing, 0); return UNITY_END(); }