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

fix(worker-rejoin-resync): task 7.5 - bound the outbound reconciliation query

Browse source 
Closes bug-1755: worker_send_resync_query() hardcoded pktlimit=0 in
its xce_job_state_query_req() call, so the controller's outbound
OPCode_jobStateQuery always sent every non-terminal job ID in one
unbounded page regardless of count - the query direction never
actually exercised the pagination mechanism task 1.4/7.4 proved
correct, unlike the report direction's real, wired
XCECD_RESYNC_REPORT_PKTLIMIT bound.

Add XCE_RESYNC_QUERY_PKTLIMIT=100, mirroring that exact precedent
(same value - a JobStateQuery entry is a bare jobid, smaller than a
JobStateReport entry), and thread it into the one call site. Existing
small-job-count behavior is unchanged; a rejoin exceeding the cap now
genuinely produces multiple wire pages, proven against the real
production call site (not a bypass) with a 250-job/3-page test.

This is the last task in worker-rejoin-resync's implementation plan.

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-02 01:23:14 +00:00
parent 613b80e991
commit 1c4cf8bd3b
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
4 changed files with 194 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

172
src/controller/testsuite/test_worker_join.c
View file

@ -14,6 +14,7 @@
#include <rtti.h>
#include <errno.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@ -1176,6 +1177,175 @@ static void test_worker_send_resync_query_sends_nothing_when_no_round(void)
"PASS\n");
}
/* ── Task 7.5: worker_send_resync_query genuinely paginates (bug-1755) ──
*
* Prior to task 7.5, worker_send_resync_query() hardcoded pktlimit=0 in its
* xce_job_state_query_req() call -- xce_job_state_query_req()'s own
* pktlimit<=0 convention (request.c) means "pktlimit = nrows", i.e. always
* exactly ONE wire page, however many jobs are queried. Task 7.4's own test
* (test_worker_resync.c, test_large_job_count_pagination_correctness) only
* proved the underlying xce_job_state_query_req() pagination MECHANISM by
* calling it directly with an explicit page-size override, bypassing this
* production call site entirely (documented live as bug-1755: "effectively
* dead code in production today because its only call site never supplies
* a bound").
*
* This test drives the REAL production call site -- worker_send_resync_
* query() itself, not a direct xce_job_state_query_req() call -- with a
* non-terminal job count on the host that genuinely exceeds the new
* XCE_RESYNC_QUERY_PKTLIMIT page cap (worker.c), and asserts the query is
* observably split across multiple wire pages: more than one packet is
* enqueued, each page's `pos` picks up exactly where the previous page's
* `count` left off, every jobid appears in exactly one page, and every
* page (including the final, deliberately partial one) carries the same
* full n_total. Req 8.1-8.3.
*/
#define TWJ_RESYNC_QUERY_PKTLIMIT 100 /* mirrors production
* XCE_RESYNC_QUERY_PKTLIMIT
* (worker.c) -- not a real
* production symbol reference (that
* #define has file-local linkage),
* kept in sync by this test's own
* page-count assertion below, which
* would fail loudly if the two ever
* drifted apart */
#define TWJ_RESYNC_QUERY_N_JOBS 250 /* > TWJ_RESYNC_QUERY_PKTLIMIT, and not
* a multiple of it, so the final wire
* page is genuinely partial (50
* entries), not just N full pages */
#define TWJ_RESYNC_QUERY_JOBID_BASE 970000ULL
#define TWJ_RESYNC_QUERY_N_PAGES \
((TWJ_RESYNC_QUERY_N_JOBS + TWJ_RESYNC_QUERY_PKTLIMIT - 1) \
/ TWJ_RESYNC_QUERY_PKTLIMIT)
static void
test_worker_send_resync_query_paginates_across_multiple_wire_pages(void)
{
setup_test_cluster();
hostattrs_t attrs;
memset(&attrs, 0, sizeof(attrs));
attrs.hosttype = makeString("LINUX");
attrs.hostmodel = makeString("x86");
attrs.cpufactor = 1.0f;
attrs.maxcpus = 4;
attrs.maxmem = 16;
attrs.maxswap = 32;
attrs.maxtmp = 64;
attrs.ndisks = 1;
worker_t* w = make_test_worker("query-host-paginate", &attrs);
TEST_ASSERT_NOT_NULL(w);
for (int i = 0; i < TWJ_RESYNC_QUERY_N_JOBS; i++) {
uint64_t jobid = TWJ_RESYNC_QUERY_JOBID_BASE + (uint64_t)i;
job_t* j = seed_test_job(jobid, JOB_STAT_RUN);
TEST_ASSERT_NOT_NULL(j);
j->job.allocs = make_test_job_alloc(jobid, "query-host-paginate");
TEST_ASSERT_NOT_NULL(j->job.allocs);
}
hostattrs_t new_attrs = attrs;
new_attrs.hosttype = makeString("LINUX");
new_attrs.hostmodel = makeString("x86");
xce_hostconfig_t hconfig;
memset(&hconfig, 0, sizeof(hconfig));
hconfig.hosttype = (char*)"!";
hconfig.hostmodel = (char*)"!";
hconfig.mxj = 4;
endpoint_t stack_ep;
memset(&stack_ep, 0, sizeof(stack_ep));
rejoin_worker(&g_tc.c, w, &stack_ep, &new_attrs, &hconfig);
TEST_ASSERT_NOT_NULL(w->resync);
TEST_ASSERT_EQUAL_INT(TWJ_RESYNC_QUERY_N_JOBS, w->resync->n_queried_jobids);
endpoint_t* ep = new_test_endpoint();
/* The REAL production call site -- no xce_job_state_query_req()
* bypass. */
TEST_ASSERT_EQUAL_INT(0, worker_send_resync_query(&g_tc.c, w, ep));
{
bool seen[TWJ_RESYNC_QUERY_N_JOBS];
int running_total = 0;
int pages = 0;
memset(seen, 0, sizeof(seen));
while (running_total < TWJ_RESYNC_QUERY_N_JOBS) {
xce_request_t decoded;
int j;
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_MESSAGE(TWJ_RESYNC_QUERY_N_JOBS,
decoded.body.jobstatequerymsg->n_total,
"every page must carry the same full n_total, not a "
"per-page total (Req 2.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(running_total,
decoded.body.jobstatequerymsg->pos,
"each page's pos must pick up exactly where the previous "
"page's count left off -- no gap, no overlap (Req 8.3)");
TEST_ASSERT_TRUE_MESSAGE(
decoded.body.jobstatequerymsg->count
<= TWJ_RESYNC_QUERY_PKTLIMIT,
"no single page may exceed the configured page cap "
"(Req 8.3)");
for (j = 0; j < decoded.body.jobstatequerymsg->count; j++) {
uint64_t jobid = decoded.body.jobstatequerymsg->jobids[j];
uint64_t idx = jobid - TWJ_RESYNC_QUERY_JOBID_BASE;
TEST_ASSERT_TRUE_MESSAGE(
idx < (uint64_t)TWJ_RESYNC_QUERY_N_JOBS,
"a jobid appearing on the wire must be one of the "
"armed round's own jobids");
TEST_ASSERT_FALSE_MESSAGE(seen[idx],
"no jobid may be sent twice across pages (Req 8.3)");
seen[idx] = true;
}
running_total += decoded.body.jobstatequerymsg->count;
pages++;
xce_free_request(&decoded);
}
TEST_ASSERT_EQUAL_INT_MESSAGE(TWJ_RESYNC_QUERY_N_PAGES, pages,
"Req 8.1-8.3: a job count exceeding the real production page "
"cap must genuinely span multiple wire pages -- observably, "
"via the REAL worker_send_resync_query() call site, not just "
"theoretically supported by xce_job_state_query_req()'s own "
"pagination loop (bug-1755, closed by task 7.5)");
TEST_ASSERT_GREATER_THAN_INT_MESSAGE(1, pages,
"a single collapsed page here would mean the hardcoded-0 "
"regression (bug-1755) is back");
for (int i = 0; i < TWJ_RESYNC_QUERY_N_JOBS; i++) {
TEST_ASSERT_TRUE_MESSAGE(seen[i],
"every armed jobid must appear exactly once across all "
"pages -- none silently dropped past a page boundary "
"(Req 8.3)");
}
}
assert_queue_empty(ep);
dereference(new_attrs.hosttype);
dereference(new_attrs.hostmodel);
endpoint_fini(ep);
free(ep);
teardown_test_cluster();
printf("test_worker_send_resync_query_paginates_across_multiple_wire_"
"pages: PASS\n");
}
/* ── Task 6.1: config push, then query, in that fixed order ──────────── */
/*
@ -1359,6 +1529,8 @@ void test_worker_join(void)
RUN_TEST(test_init_worker_arms_reconciliation_round_when_jobs_exist);
RUN_TEST(test_worker_send_resync_query_sends_armed_round);
RUN_TEST(test_worker_send_resync_query_sends_nothing_when_no_round);
RUN_TEST(
test_worker_send_resync_query_paginates_across_multiple_wire_pages);
RUN_TEST(test_config_push_then_resync_query_wire_order);
RUN_TEST(test_worker_push_config_carries_configured_resync_window);
}

18
src/controller/worker.c
View file

@ -663,6 +663,21 @@ static int worker_push_config(cluster_t* c, endpoint_t* ep)
return rc;
}
/*
* worker-rejoin-resync task 7.5 (bug-1755, Req 8.1-8.3): page-size cap for
* this controller's own OPCode_jobStateQuery sends, mirroring the worker's
* own already-wired XCECD_RESYNC_REPORT_PKTLIMIT (controller_connector.c)
* for the report direction -- same constant-driven approach, applied here
* to the query direction so xce_job_state_query_req()'s already-implemented
* pagination loop (task 1.4) is genuinely exercised in production instead
* of always collapsing to a single unbounded page (bug-1755's finding). A
* JobStateQuery entry (bare jobid) is smaller on the wire than a
* JobStateReport entry (jobid + status + exitCode + finishTimeMs +
* queried), so the query direction can safely use the same page size the
* report direction already proves safe, without inventing a smaller one.
*/
#define XCE_RESYNC_QUERY_PKTLIMIT 100
/*
* worker_send_resync_query — worker-rejoin-resync task 6.1 (Req 1.1-1.3,
* 2.1, 2.2): sends the round's job-state query to a worker for which
@ -727,7 +742,8 @@ static int worker_send_resync_query(cluster_t* c, worker_t* w, endpoint_t* ep)
query.n_total = n;
query.jobids = ids;
xce_job_state_query_req(ep, c->myhost, &query, NULL, NULL, 0);
xce_job_state_query_req(ep, c->myhost, &query, NULL, NULL,
XCE_RESYNC_QUERY_PKTLIMIT);
free(ids);