diff --git a/.kiro/specs/worker-rejoin-resync/tasks.md b/.kiro/specs/worker-rejoin-resync/tasks.md index 74310a32..536ffb0f 100644 --- a/.kiro/specs/worker-rejoin-resync/tasks.md +++ b/.kiro/specs/worker-rejoin-resync/tasks.md @@ -21,7 +21,7 @@ - _Boundary: worker test infrastructure_ - _Note: satisfied by pre-existing infrastructure — see Implementation Notes._ -- [ ] 1.4 (P) Add wire encode/decode helpers for the reconciliation exchange +- [x] 1.4 (P) Add wire encode/decode helpers for the reconciliation exchange - Add request/reply encode and decode helpers for the new query and report messages, following the existing paginated-list pattern already used for job/worker/resource listings - Observable completion: a round-trip encode-then-decode of a query and of a report (including an empty page and a multi-page case) reproduces the original data exactly - _Requirements: 8.1, 8.2, 8.3_ diff --git a/src/include/request.h b/src/include/request.h index 7d4fade6..a39ebefd 100644 --- a/src/include/request.h +++ b/src/include/request.h @@ -60,6 +60,31 @@ extern int xce_ctrlstatus_req(endpoint_t*); extern int xce_get_req(endpoint_t*, xce_get_t*); extern int xce_reply_get(endpoint_t*, xce_peerinfo_t*, xce_reply_t*, xce_check_fn_t, void*, int); + +/* + * xce_job_state_query_req / xce_job_state_report_req: send the worker-rejoin- + * resync job-state reconciliation exchange's paginated request/reply pair + * (worker-rejoin-resync task 1.4, Req 8.1/8.2/8.3) -- + * OPCode_jobStateQuery (controller -> worker, query->jobids) and + * OPCode_jobStateReport (worker -> controller, report->entries) -- using the + * same {pos,count,ntotal} pagination idiom as xce_push_data_req()/ + * xce_reply_get() above: `pktlimit <= 0` sends the whole list in one page; + * a zero-length list still sends exactly one page with count == 0 (an empty + * page is a valid, complete exchange, not something to skip). The receiving + * side decodes each page with the generic xce_decode() -- these opcodes are + * their own dedicated Request body union arms (jobstatequerymsg/ + * jobstatereportmsg, added by task 1.1's schema), not multiplexed through + * xce_data_t, so encode/decode of each page's payload is already handled by + * the generated encode_JobStateQuery()/encode_JobStateReport() (controller.capnp.c) + * that xce_encode()/xce_decode() call into. + */ +extern int xce_job_state_query_req(endpoint_t*, xce_peerinfo_t*, + xce_job_state_query_t*, + xce_check_fn_t, void*, int); +extern int xce_job_state_report_req(endpoint_t*, xce_peerinfo_t*, + xce_job_state_report_t*, + xce_check_fn_t, void*, int); + extern int xce_notify_req(endpoint_t*, xce_notify_data_t*); extern int extract_data(dataset_t*, xce_data_t*); extern int xce_update_req(endpoint_t*, xce_update_data_t*); diff --git a/src/lib/meson.build b/src/lib/meson.build index 4057e4ec..65f5072c 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -167,6 +167,7 @@ libxce_test_src = ['testsuite/test.c', 'testsuite/test_wobj.c', 'testsuite/test_qobj.c', 'testsuite/test_request.c', + 'testsuite/test_job_state_wire.c', 'testsuite/test_cbuf.c', 'testsuite/test_zarray_cow.c', 'testsuite/test_kvmgr_capture.c', diff --git a/src/lib/request.c b/src/lib/request.c index f3b972e7..d2063365 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -1783,6 +1783,149 @@ int xce_reply_get(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply, return rc; } +/* + * xce_job_state_query_req: send OPCode_jobStateQuery (controller -> worker), + * paginated over `query->jobids` using the {pos,count,ntotal} pagination + * idiom established by xce_push_data_req()/xce_reply_get() above + * (worker-rejoin-resync task 1.4, Req 8.1/8.2/8.3). Unlike xce_push_data_req() + * (which multiplexes several XceData_data_* tables through one OPCode_push / + * xce_data_t container), JobStateQuery is its own dedicated Request body + * union arm (jobstatequerymsg, added by task 1.1's schema) -- so each page is + * sent directly as its own xce_request_t, with no xce_data_t indirection and + * no heap container to free afterward: `page` below only ever borrows a + * slice of the caller's own query->jobids array (mirrors how + * xce_push_data_req's data1.data.jobs->jobs slices the caller's jobs array), + * never owns it -- encode_JobStateQuery() (controller.capnp.c, generated) + * only reads it while building the wire message, so there is nothing to + * release once xce_request() returns. + * + * A zero-job query (query->n_total == 0) still sends exactly one page with + * count == 0, matching xce_push_data_req/xce_reply_get's do/while shape -- + * "no jobs to query" is itself a valid, complete exchange, not something to + * skip sending (Req 8.1). + */ +int xce_job_state_query_req(endpoint_t* ep, xce_peerinfo_t* from, + xce_job_state_query_t* query, + xce_check_fn_t checkf, void* checkarg, int pktlimit) +{ + int rc = 0; + int pos = 0; + int nrows = 0; + int checkstatus = 0; + + if ((ep == NULL) || (from == NULL) || (query == NULL)) { + return -EINVAL; + } + + nrows = query->n_total; + + if (pktlimit <= 0) { + pktlimit = nrows; + } + + do { + int nc = nrows - pos; + xce_job_state_query_t page; + xce_request_t request; + + if (nc > pktlimit) { + nc = pktlimit; + } + + if (checkf != NULL) { + checkstatus = (*checkf)(checkarg); + + if (checkstatus < 0) { + nc = 0; + } + } + + page.n_total = query->n_total; + page.pos = pos; + page.count = nc; + page.jobids = (nc == 0) ? NULL : &(query->jobids[pos]); + + memset(&request, 0, sizeof(request)); + request.opcode = OPCode_jobStateQuery; + request.body_which = Request_body_jobstatequerymsg; + request.body.jobstatequerymsg = &page; + + rc = xce_request(ep, &request); + if (rc < 0) { + break; + } + + pos += nc; + } while ((pos < nrows) && (checkstatus >= 0)); + + return rc; +} + +/* + * xce_job_state_report_req: send OPCode_jobStateReport (worker -> controller), + * paginated over `report->entries` using the same {pos,count,ntotal} idiom + * (Req 8.1/8.2/8.3). Companion to xce_job_state_query_req() above -- same + * borrow-not-own `page` shape (entries slices report->entries, an array of + * xce_job_state_entry_t* the caller owns), same "an empty report is still + * one sent page" rule. + */ +int xce_job_state_report_req(endpoint_t* ep, xce_peerinfo_t* from, + xce_job_state_report_t* report, + xce_check_fn_t checkf, void* checkarg, int pktlimit) +{ + int rc = 0; + int pos = 0; + int nrows = 0; + int checkstatus = 0; + + if ((ep == NULL) || (from == NULL) || (report == NULL)) { + return -EINVAL; + } + + nrows = report->n_total; + + if (pktlimit <= 0) { + pktlimit = nrows; + } + + do { + int nc = nrows - pos; + xce_job_state_report_t page; + xce_request_t request; + + if (nc > pktlimit) { + nc = pktlimit; + } + + if (checkf != NULL) { + checkstatus = (*checkf)(checkarg); + + if (checkstatus < 0) { + nc = 0; + } + } + + page.n_total = report->n_total; + page.pos = pos; + page.count = nc; + page.entries = (nc == 0) ? NULL : &(report->entries[pos]); + + memset(&request, 0, sizeof(request)); + request.opcode = OPCode_jobStateReport; + request.body_which = Request_body_jobstatereportmsg; + request.body.jobstatereportmsg = &page; + + rc = xce_request(ep, &request); + if (rc < 0) { + break; + } + + pos += nc; + } while ((pos < nrows) && (checkstatus >= 0)); + + return rc; +} + int extract_data(dataset_t* dataset, xce_data_t* getdata) { int ntotal; diff --git a/src/lib/testsuite/test.c b/src/lib/testsuite/test.c index d29c7171..4d4976b1 100644 --- a/src/lib/testsuite/test.c +++ b/src/lib/testsuite/test.c @@ -30,6 +30,7 @@ extern void test_jobj_hardening(void); extern void test_qobj(void); extern void test_wobj(void); extern void test_request(void); +extern void test_job_state_wire(void); extern void test_cbuf(void); extern void test_zarray_cow(void); extern void test_kvmgr_capture(void); @@ -152,6 +153,7 @@ int main(int argc, char* argv[]) TEST(test_wobj); TEST(test_qobj); TEST(test_request); + TEST(test_job_state_wire); TEST(test_cbuf); TEST(test_zarray_cow); TEST(test_kvmgr_capture); diff --git a/src/lib/testsuite/test_job_state_wire.c b/src/lib/testsuite/test_job_state_wire.c new file mode 100644 index 00000000..8be50611 --- /dev/null +++ b/src/lib/testsuite/test_job_state_wire.c @@ -0,0 +1,409 @@ +/* + * test_job_state_wire.c -- worker-rejoin-resync task 1.4: wire encode/decode + * round-trip tests for the job-state reconciliation exchange + * (OPCode_jobStateQuery / OPCode_jobStateReport), Req 8.1/8.2/8.3. + * + * Mirrors test_request.c's HA-heartbeat round-trip style: there is no live + * socket in these unit tests, so a real (heap) endpoint_t is built via + * endpoint_get() -- NOT a hand-rolled stub struct (bug-378: an + * under-sized/wrong-typed stand-in for endpoint_t caused a global-buffer- + * overflow when passed where a real endpoint_t* was expected). endpoint_enq() + * only queues onto ep->outq (a plain in-memory linked list, dataq_enq()) and + * skips the iomgr poll-arm step entirely when ep->iomgr == NULL (the default + * after endpoint_get()), so xce_job_state_query_req()/xce_job_state_report_req() + * can be exercised end-to-end -- real xce_encode() on send, real xce_decode() + * on "receive" -- without a socket, exactly as + * test_xce_request_propagates_send_failure() (test_request_send_propagation.c) + * already does for a single OPCode_nop request. Each queued page is popped + * with endpoint_deq()/ITPTR_TO_PACKET() and decoded with xce_decode(), which + * takes ownership of (frees) the packet -- matching test_request.c's + * documented xce_decode() contract. + */ + +#include +#include +#include +#include +#include "netio.h" +#include "request.h" +#include "xce/types.h" + +static endpoint_t* new_test_endpoint(void) +{ + endpoint_t* ep = endpoint_get(); + TEST_ASSERT_NOT_NULL(ep); + return ep; +} + +/* Pop one queued page off ep->outq and decode it. Fails the test if the + * queue is empty. Caller must xce_free_request() the result. */ +static void pop_and_decode(endpoint_t* ep, xce_request_t* out) +{ + data_item_t* it = endpoint_deq(ep); + TEST_ASSERT_NOT_NULL(it); + + io_packet_t* pkt = ITPTR_TO_PACKET(it); + + memset(out, 0, sizeof(*out)); + TEST_ASSERT_EQUAL_INT(0, xce_decode(pkt, out)); /* consumes/frees pkt */ +} + +static void assert_queue_empty(endpoint_t* ep) +{ + TEST_ASSERT_NULL(endpoint_deq(ep)); +} + +/* ── OPCode_jobStateQuery round trips (controller -> worker) ────────────── */ + +static void test_job_state_query_req_null_args(void) +{ + xce_job_state_query_t query = { 0 }; + char nm[] = "tester"; + char ky[] = "key"; + xce_peerinfo_t from = { .name = nm, .key = ky }; + + TEST_ASSERT_EQUAL_INT(-EINVAL, + xce_job_state_query_req(NULL, &from, &query, NULL, NULL, 0)); + + endpoint_t* ep = new_test_endpoint(); + TEST_ASSERT_EQUAL_INT(-EINVAL, + xce_job_state_query_req(ep, NULL, &query, NULL, NULL, 0)); + TEST_ASSERT_EQUAL_INT(-EINVAL, + xce_job_state_query_req(ep, &from, NULL, NULL, NULL, 0)); + + endpoint_fini(ep); + free(ep); + printf("test_job_state_query_req_null_args: PASS\n"); +} + +static void test_job_state_query_roundtrip_normal(void) +{ + uint64_t jobids[5] = { 101, 202, 303, 404, 505 }; + xce_job_state_query_t query = { + .n_total = 5, .pos = 0, .count = 5, .jobids = jobids + }; + char nm[] = "tester"; + char ky[] = "key"; + xce_peerinfo_t from = { .name = nm, .key = ky }; + + endpoint_t* ep = new_test_endpoint(); + + /* pktlimit <= 0 -> whole list in one page. */ + TEST_ASSERT_EQUAL_INT(0, + xce_job_state_query_req(ep, &from, &query, NULL, NULL, 0)); + + xce_request_t decoded; + pop_and_decode(ep, &decoded); + + TEST_ASSERT_EQUAL_INT(OPCode_jobStateQuery, decoded.opcode); + TEST_ASSERT_EQUAL_INT(Request_body_jobstatequerymsg, decoded.body_which); + TEST_ASSERT_NOT_NULL(decoded.body.jobstatequerymsg); + TEST_ASSERT_EQUAL_INT(5, decoded.body.jobstatequerymsg->n_total); + TEST_ASSERT_EQUAL_INT(0, decoded.body.jobstatequerymsg->pos); + TEST_ASSERT_EQUAL_INT(5, decoded.body.jobstatequerymsg->count); + TEST_ASSERT_NOT_NULL(decoded.body.jobstatequerymsg->jobids); + for (int i = 0; i < 5; i++) { + TEST_ASSERT_EQUAL_UINT64(jobids[i], decoded.body.jobstatequerymsg->jobids[i]); + } + + xce_free_request(&decoded); + assert_queue_empty(ep); + + endpoint_fini(ep); + free(ep); + printf("test_job_state_query_roundtrip_normal: PASS\n"); +} + +static void test_job_state_query_roundtrip_empty(void) +{ + xce_job_state_query_t query = { + .n_total = 0, .pos = 0, .count = 0, .jobids = NULL + }; + char nm[] = "tester"; + char ky[] = "key"; + xce_peerinfo_t from = { .name = nm, .key = ky }; + + endpoint_t* ep = new_test_endpoint(); + + TEST_ASSERT_EQUAL_INT(0, + xce_job_state_query_req(ep, &from, &query, NULL, NULL, 0)); + + /* Req 8.1: zero jobs to query is still a complete, single-page exchange + * -- not skipped, not left unsent. */ + xce_request_t decoded; + pop_and_decode(ep, &decoded); + + TEST_ASSERT_EQUAL_INT(OPCode_jobStateQuery, decoded.opcode); + TEST_ASSERT_EQUAL_INT(Request_body_jobstatequerymsg, decoded.body_which); + TEST_ASSERT_NOT_NULL(decoded.body.jobstatequerymsg); + TEST_ASSERT_EQUAL_INT(0, decoded.body.jobstatequerymsg->n_total); + TEST_ASSERT_EQUAL_INT(0, decoded.body.jobstatequerymsg->pos); + TEST_ASSERT_EQUAL_INT(0, decoded.body.jobstatequerymsg->count); + TEST_ASSERT_NULL(decoded.body.jobstatequerymsg->jobids); + + xce_free_request(&decoded); + assert_queue_empty(ep); + + endpoint_fini(ep); + free(ep); + printf("test_job_state_query_roundtrip_empty: PASS\n"); +} + +static void test_job_state_query_roundtrip_multipage(void) +{ + uint64_t jobids[7] = { 1, 2, 3, 4, 5, 6, 7 }; + xce_job_state_query_t query = { + .n_total = 7, .pos = 0, .count = 7, .jobids = jobids + }; + char nm[] = "tester"; + char ky[] = "key"; + xce_peerinfo_t from = { .name = nm, .key = ky }; + + endpoint_t* ep = new_test_endpoint(); + + /* pktlimit=3 over 7 jobids -> pages of 3, 3, 1 (Req 8.1/8.3: nothing + * truncated or dropped across pages). */ + TEST_ASSERT_EQUAL_INT(0, + xce_job_state_query_req(ep, &from, &query, NULL, NULL, 3)); + + uint64_t rebuilt[7]; + int rebuilt_n = 0; + int expected_counts[3] = { 3, 3, 1 }; + int expected_pos[3] = { 0, 3, 6 }; + + for (int page = 0; page < 3; page++) { + xce_request_t decoded; + pop_and_decode(ep, &decoded); + + TEST_ASSERT_EQUAL_INT(OPCode_jobStateQuery, decoded.opcode); + TEST_ASSERT_EQUAL_INT(Request_body_jobstatequerymsg, decoded.body_which); + TEST_ASSERT_NOT_NULL(decoded.body.jobstatequerymsg); + TEST_ASSERT_EQUAL_INT(7, decoded.body.jobstatequerymsg->n_total); + TEST_ASSERT_EQUAL_INT(expected_pos[page], decoded.body.jobstatequerymsg->pos); + TEST_ASSERT_EQUAL_INT(expected_counts[page], decoded.body.jobstatequerymsg->count); + + for (int i = 0; i < decoded.body.jobstatequerymsg->count; i++) { + rebuilt[rebuilt_n++] = decoded.body.jobstatequerymsg->jobids[i]; + } + + xce_free_request(&decoded); + } + + assert_queue_empty(ep); + + TEST_ASSERT_EQUAL_INT(7, rebuilt_n); + for (int i = 0; i < 7; i++) { + TEST_ASSERT_EQUAL_UINT64(jobids[i], rebuilt[i]); + } + + endpoint_fini(ep); + free(ep); + printf("test_job_state_query_roundtrip_multipage: PASS\n"); +} + +/* ── OPCode_jobStateReport round trips (worker -> controller) ───────────── */ + +static void test_job_state_report_req_null_args(void) +{ + xce_job_state_report_t report = { 0 }; + char nm[] = "tester"; + char ky[] = "key"; + xce_peerinfo_t from = { .name = nm, .key = ky }; + + TEST_ASSERT_EQUAL_INT(-EINVAL, + xce_job_state_report_req(NULL, &from, &report, NULL, NULL, 0)); + + endpoint_t* ep = new_test_endpoint(); + TEST_ASSERT_EQUAL_INT(-EINVAL, + xce_job_state_report_req(ep, NULL, &report, NULL, NULL, 0)); + TEST_ASSERT_EQUAL_INT(-EINVAL, + xce_job_state_report_req(ep, &from, NULL, NULL, NULL, 0)); + + endpoint_fini(ep); + free(ep); + printf("test_job_state_report_req_null_args: PASS\n"); +} + +/* Covers all four JobStateStatus values and both queried true/false, plus + * exitCode/finishTimeMs survival for finished entries. */ +static void test_job_state_report_roundtrip_normal(void) +{ + xce_job_state_entry_t e0 = { + .jobid = 9001, .status = JobStateStatus_running, + .exit_code = 0, .finish_time_ms = 0, .queried = 1 + }; + xce_job_state_entry_t e1 = { + .jobid = 9002, .status = JobStateStatus_finishedOk, + .exit_code = 0, .finish_time_ms = 1693500000123ULL, .queried = 0 + }; + xce_job_state_entry_t e2 = { + .jobid = 9003, .status = JobStateStatus_finishedFail, + .exit_code = 137, .finish_time_ms = 1693500005555ULL, .queried = 1 + }; + xce_job_state_entry_t e3 = { + .jobid = 9004, .status = JobStateStatus_unrecognized, + .exit_code = 0, .finish_time_ms = 0, .queried = 0 + }; + xce_job_state_entry_t* entries[4] = { &e0, &e1, &e2, &e3 }; + xce_job_state_report_t report = { + .n_total = 4, .pos = 0, .count = 4, .entries = entries + }; + char nm[] = "tester"; + char ky[] = "key"; + xce_peerinfo_t from = { .name = nm, .key = ky }; + + endpoint_t* ep = new_test_endpoint(); + + TEST_ASSERT_EQUAL_INT(0, + xce_job_state_report_req(ep, &from, &report, NULL, NULL, 0)); + + xce_request_t decoded; + pop_and_decode(ep, &decoded); + + TEST_ASSERT_EQUAL_INT(OPCode_jobStateReport, decoded.opcode); + TEST_ASSERT_EQUAL_INT(Request_body_jobstatereportmsg, decoded.body_which); + TEST_ASSERT_NOT_NULL(decoded.body.jobstatereportmsg); + TEST_ASSERT_EQUAL_INT(4, decoded.body.jobstatereportmsg->n_total); + TEST_ASSERT_EQUAL_INT(0, decoded.body.jobstatereportmsg->pos); + TEST_ASSERT_EQUAL_INT(4, decoded.body.jobstatereportmsg->count); + TEST_ASSERT_NOT_NULL(decoded.body.jobstatereportmsg->entries); + + xce_job_state_entry_t** got = decoded.body.jobstatereportmsg->entries; + + TEST_ASSERT_EQUAL_UINT64(9001, got[0]->jobid); + TEST_ASSERT_EQUAL_INT(JobStateStatus_running, got[0]->status); + TEST_ASSERT_EQUAL_INT(1, got[0]->queried); + + TEST_ASSERT_EQUAL_UINT64(9002, got[1]->jobid); + TEST_ASSERT_EQUAL_INT(JobStateStatus_finishedOk, got[1]->status); + TEST_ASSERT_EQUAL_INT(0, got[1]->queried); + TEST_ASSERT_EQUAL_UINT64(1693500000123ULL, got[1]->finish_time_ms); + + TEST_ASSERT_EQUAL_UINT64(9003, got[2]->jobid); + TEST_ASSERT_EQUAL_INT(JobStateStatus_finishedFail, got[2]->status); + TEST_ASSERT_EQUAL_INT(1, got[2]->queried); + TEST_ASSERT_EQUAL_INT32(137, got[2]->exit_code); + TEST_ASSERT_EQUAL_UINT64(1693500005555ULL, got[2]->finish_time_ms); + + TEST_ASSERT_EQUAL_UINT64(9004, got[3]->jobid); + TEST_ASSERT_EQUAL_INT(JobStateStatus_unrecognized, got[3]->status); + TEST_ASSERT_EQUAL_INT(0, got[3]->queried); + + xce_free_request(&decoded); + assert_queue_empty(ep); + + endpoint_fini(ep); + free(ep); + printf("test_job_state_report_roundtrip_normal: PASS\n"); +} + +static void test_job_state_report_roundtrip_empty(void) +{ + xce_job_state_report_t report = { + .n_total = 0, .pos = 0, .count = 0, .entries = NULL + }; + char nm[] = "tester"; + char ky[] = "key"; + xce_peerinfo_t from = { .name = nm, .key = ky }; + + endpoint_t* ep = new_test_endpoint(); + + TEST_ASSERT_EQUAL_INT(0, + xce_job_state_report_req(ep, &from, &report, NULL, NULL, 0)); + + xce_request_t decoded; + pop_and_decode(ep, &decoded); + + TEST_ASSERT_EQUAL_INT(OPCode_jobStateReport, decoded.opcode); + TEST_ASSERT_EQUAL_INT(Request_body_jobstatereportmsg, decoded.body_which); + TEST_ASSERT_NOT_NULL(decoded.body.jobstatereportmsg); + TEST_ASSERT_EQUAL_INT(0, decoded.body.jobstatereportmsg->n_total); + TEST_ASSERT_EQUAL_INT(0, decoded.body.jobstatereportmsg->pos); + TEST_ASSERT_EQUAL_INT(0, decoded.body.jobstatereportmsg->count); + TEST_ASSERT_NULL(decoded.body.jobstatereportmsg->entries); + + xce_free_request(&decoded); + assert_queue_empty(ep); + + endpoint_fini(ep); + free(ep); + printf("test_job_state_report_roundtrip_empty: PASS\n"); +} + +static void test_job_state_report_roundtrip_multipage(void) +{ + xce_job_state_entry_t e[5]; + xce_job_state_entry_t* entries[5]; + for (int i = 0; i < 5; i++) { + memset(&e[i], 0, sizeof(e[i])); + e[i].jobid = 8000 + (uint64_t)i; + e[i].status = JobStateStatus_finishedOk; + e[i].finish_time_ms = 1700000000000ULL + (uint64_t)i; + e[i].queried = (i % 2 == 0) ? 1 : 0; + entries[i] = &e[i]; + } + xce_job_state_report_t report = { + .n_total = 5, .pos = 0, .count = 5, .entries = entries + }; + char nm[] = "tester"; + char ky[] = "key"; + xce_peerinfo_t from = { .name = nm, .key = ky }; + + endpoint_t* ep = new_test_endpoint(); + + /* pktlimit=2 over 5 entries -> pages of 2, 2, 1. */ + TEST_ASSERT_EQUAL_INT(0, + xce_job_state_report_req(ep, &from, &report, NULL, NULL, 2)); + + uint64_t rebuilt_jobids[5]; + int rebuilt_queried[5]; + int rebuilt_n = 0; + int expected_counts[3] = { 2, 2, 1 }; + int expected_pos[3] = { 0, 2, 4 }; + + for (int page = 0; page < 3; page++) { + xce_request_t decoded; + pop_and_decode(ep, &decoded); + + TEST_ASSERT_EQUAL_INT(OPCode_jobStateReport, decoded.opcode); + TEST_ASSERT_EQUAL_INT(Request_body_jobstatereportmsg, decoded.body_which); + TEST_ASSERT_NOT_NULL(decoded.body.jobstatereportmsg); + TEST_ASSERT_EQUAL_INT(5, decoded.body.jobstatereportmsg->n_total); + TEST_ASSERT_EQUAL_INT(expected_pos[page], decoded.body.jobstatereportmsg->pos); + TEST_ASSERT_EQUAL_INT(expected_counts[page], decoded.body.jobstatereportmsg->count); + + for (int i = 0; i < decoded.body.jobstatereportmsg->count; i++) { + xce_job_state_entry_t* got = decoded.body.jobstatereportmsg->entries[i]; + rebuilt_jobids[rebuilt_n] = got->jobid; + rebuilt_queried[rebuilt_n] = got->queried; + rebuilt_n++; + } + + xce_free_request(&decoded); + } + + assert_queue_empty(ep); + + TEST_ASSERT_EQUAL_INT(5, rebuilt_n); + for (int i = 0; i < 5; i++) { + TEST_ASSERT_EQUAL_UINT64(e[i].jobid, rebuilt_jobids[i]); + TEST_ASSERT_EQUAL_INT(e[i].queried, rebuilt_queried[i]); + } + + endpoint_fini(ep); + free(ep); + printf("test_job_state_report_roundtrip_multipage: PASS\n"); +} + +void test_job_state_wire(void) +{ + printf("\n>> job-state reconciliation wire codec round-trip tests\n\n"); + RUN_TEST(test_job_state_query_req_null_args); + RUN_TEST(test_job_state_query_roundtrip_normal); + RUN_TEST(test_job_state_query_roundtrip_empty); + RUN_TEST(test_job_state_query_roundtrip_multipage); + RUN_TEST(test_job_state_report_req_null_args); + RUN_TEST(test_job_state_report_roundtrip_normal); + RUN_TEST(test_job_state_report_roundtrip_empty); + RUN_TEST(test_job_state_report_roundtrip_multipage); +}