lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(job-array-single-wal-record): wire dispatcher was silently dropping compact array pushes

Browse source 
extract_data() (src/lib/request.c), the shared reassembly step every
inbound push passes through inside vtask_entry() before any receiver's
handler runs, had no case for Data_data_jobArray -- it hit
default: rc=-EINVAL. Because vtask_entry()'s callnext check treats any
nonzero return (including a negative errno) as "reassembly complete,
proceed," this never surfaced as an error: the real payload was freed
and the receiver silently handed an empty, zeroed dataset_t, with no
log anywhere in the chain. Tasks 5.1-5.3 were each individually
correct and tested in isolation, but the feature was dead on the real
wire end-to-end.

Adds case Data_data_jobArray: passing the single, unchunked
descriptor straight through (no pagination needed, unlike the
existing LIST-shaped cases) and nulls the source pointer after
transfer -- verified this is not cosmetic, since free_Request()'s
pushmsg case would otherwise double-free it via free_Data()'s live
Data_data_jobArray branch.

Proven with a real xce_encode()/xce_decode() round-trip through the
fixed extract_data() -> sched_handle_dataset() -> sched_on_job_array()
(task 5.3, unmodified) chain, not just a unit-level check on
extract_data() alone. Also confirms task 5.4's future API-server
receive path shares this exact fix, since both go through the same
xcecli_ctx_init() -> vtask_entry() -> extract_data() pipeline.

One self-flagged, non-blocking residual gap (pre-existing since task
5.3, out of this task's request.c-only boundary): dataset->rows for
the jobArray case is never deep-freed downstream, leaking one bounded
xce_job_submit_t per successfully-delivered array push commit.

Task 5.3b of job-array-single-wal-record (Requirements 6.1, 6.2, 6.4, 8.4).
Inserted mid-run after independent review confirmed the gap; user
confirmed sequencing (fix now, before task 5.4).
This commit is contained in:
Rongsong Shen 2026-07-15 10:59:39 +00:00
parent d16847185e
commit 1cbd7df1a8
4 changed files with 216 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

97
src/scheduler/testsuite/test_job_array_expand.c
View file

@ -248,6 +248,102 @@ static void test_job_array_null_descriptor_returns_einval(void)
TEST_ASSERT_EQUAL_INT(-EINVAL, sched_on_job_array(&sched, NULL));
}
/* ── job-array-single-wal-record task 5.3b: full wire-dispatch chain ──────
* 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-expand chain is intact end to end, not just sched_on_job_array()
* in isolation. */
static void test_wire_dispatch_chain_delivers_n_tasks(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_t **jobs = NULL;
int njobs;
int i;
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 = Data_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: N schedulable tasks, correctly addressed via base_jobid + i. */
njobs = pop_job_update(&sched, &jobs);
TEST_ASSERT_EQUAL_INT(4, njobs);
TEST_ASSERT_EQUAL_UINT64(8000, jobs[0]->jobid);
TEST_ASSERT_EQUAL_UINT64(8001, jobs[1]->jobid);
TEST_ASSERT_EQUAL_UINT64(8002, jobs[2]->jobid);
TEST_ASSERT_EQUAL_UINT64(8003, jobs[3]->jobid);
TEST_ASSERT_EQUAL_STRING("normal", jobs[0]->submit->queue);
TEST_ASSERT_EQUAL_STRING("normal", jobs[3]->submit->queue);
for (i = 0; i < njobs; i++) {
free_Job_ptr(&jobs[i]);
}
free(jobs);
/* 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
* deliberately does not null dataset.rows out (every element is its own
* clone -- see that function's own comment), so this test, as the
* descriptor's owner, frees it directly. decoded.body.pushmsg is now a
* hollow container, safe to free via the normal request teardown. */
xce_free_request(&decoded);
free_JobSubmit_ptr((xce_job_submit_t **)&dataset.rows);
}
int main(int argc, char* argv[])
{
(void)argc; (void)argv;
@ -257,5 +353,6 @@ int main(int argc, char* argv[])
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_n_tasks, 0);
return UNITY_END();
}