feat(worker-rejoin-resync): add reconciliation wire codec helpers (task 1.4)

Add xce_job_state_query_req()/xce_job_state_report_req() to request.c,
paginating job IDs / job-state entries across OPCode_jobStateQuery /
OPCode_jobStateReport messages via direct Request.body union arms
(mirroring xce_jobdispatch_req/xce_notify_req's direct-send shape, not
xce_push_data_req's XceData-table multiplexing -- these are dedicated
union arms, not multiplexed push/reply-get traffic). 8 round-trip tests
(normal/empty/multi-page for both query and report, covering all 4
JobStateStatus values and both queried states) all pass.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Y4M5MJM2zK6nU43da267Ku
This commit is contained in:
Rongsong Shen 2026-09-01 09:34:25 +00:00
parent 9fbbd12b36
commit 9f3dfe7903
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
6 changed files with 581 additions and 1 deletions

View file

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