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

test(worker-rejoin-resync): task 7.4 - large job-count pagination correctness

Browse source 
Add test_large_job_count_pagination_correctness, proving both
OPCode_jobStateQuery and OPCode_jobStateReport genuinely span many
wire pages (1042 jobs, 11 pages each direction, including a partial
final page) with zero jobs dropped/duplicated/truncated - exact-count
assertions throughout (page continuity, no-duplicate bitmap, exact WAL
delta, exact hook-fire count), extending task 6.2's 2-page proof by an
order of magnitude. Test-only, no production changes.

Logs bug-1755: the underlying query-pagination mechanism this test
proves correct is never exercised in production - worker_send_resync_
query() hardcodes pktlimit=0, so the controller's real query send
always produces one unbounded page regardless of job count, asymmetric
with the report direction's real XCECD_RESYNC_REPORT_PKTLIMIT bound.
No hard transport-layer size cap exists, so this isn't a demonstrated
failure mode, but it contradicts design.md's explicit "both paginated"
claim in this feature's own headline mechanism. Adds task 7.5 to close
it with a small, well-scoped fix before the feature-level gate, rather
than leaving a silently-false design claim in the completed feature.

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 00:48:15 +00:00
parent 165fc1586a
commit 613b80e991
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
3 changed files with 715 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

679
src/controller/testsuite/test_worker_resync.c
View file

@ -4067,6 +4067,684 @@ static void test_burst_recovery_unbounded_by_job_count(void)
printf("test_burst_recovery_unbounded_by_job_count: PASS\n");
}
/* ── Task 7.4: Large job-count pagination correctness (Req 8.1-8.3) ──────
*
* design.md's own "Performance/Load" testing-strategy line: "Reconciliation
* exchange for a worker running an order of magnitude more jobs than a
* single message can carry completes correctly across multiple pages, not
* truncated (Requirement 8.1-8.3)." Tasks 7.1-7.3 all deliberately stayed
* within ONE wire page each (task 7.3's own Implementation Notes:
* "Deliberately kept to a single wire page (page-boundary correctness is
* task 7.4's separate concern)") -- THIS is that separate concern: proving
* BOTH directions of the exchange (controller->worker OPCode_jobStateQuery,
* worker->controller OPCode_jobStateReport) genuinely span MULTIPLE wire
* pages for a large job count, and that resync_handle_report_page()'s
* accumulator (task 6.2) reassembles all of them into exactly one applied
* report -- not merely re-proving task 7.1's four-outcome/task 7.3's burst
* recovery logic at a bigger N.
*
* Real production page-size constants (found by reading the actual call
* sites, not assumed):
* - REPORT direction (worker -> controller): xcecd's real production call
* site (controller_connector.c's handle_jobstatequery ->
* xcecd_resync_handle_query_page -> xcecd_resync_answer_query) passes
* XCECD_RESYNC_REPORT_PKTLIMIT == 100 (controller_connector.c) into
* xce_job_state_report_req() (task 1.4, request.c) -- a REAL, wired,
* production page cap.
* - QUERY direction (controller -> worker): the controller's own real
* production call site, worker_send_resync_query() (worker.c:730),
* hardcodes pktlimit=0 in its xce_job_state_query_req() call --
* xce_job_state_query_req()'s own pktlimit<=0 convention means
* "pktlimit = nrows", i.e. ALWAYS exactly ONE wire page, however many
* jobs are queried. There is no real production constant to cite for
* this direction -- see this task's own CONCERNS (bug-1755). To
* genuinely exercise the QUERY side's already-implemented pagination
* loop (task 1.4) at scale, this test calls xce_job_state_query_req()
* directly (bypassing worker_send_resync_query()'s hardcoded 0),
* supplying the SAME page size as the worker's own real report cap for
* symmetry -- the exact same production function, exercised with a
* real bound instead of an effectively-disabled one.
*
* TWR_LARGE_N_TOTAL (below) is an order of magnitude larger than
* TWR_LARGE_PKTLIMIT (100), yielding a double-digit page count in EACH
* direction, with a genuinely PARTIAL final page (the total, and each of
* the four outcome-group sizes, are deliberately NOT multiples of the page
* limit) -- so several wire pages straddle two different outcome groups, a
* genuine page-boundary-falls-mid-group case, and the very last page of
* each direction is short. Covers all four outcome kinds (running,
* finished-success, finished-failure, unrecognized) so this is not a
* degenerate single-outcome test.
*/
#define TWR_JOBID_LARGE_BASE 900000ULL
#define TWR_LARGE_N_RUNNING 260
#define TWR_LARGE_N_FINOK 245
#define TWR_LARGE_N_FINFAIL 235
#define TWR_LARGE_N_UNREC 302
#define TWR_LARGE_N_TOTAL 1042 /* 260+245+235+302 -- not a multiple of
* TWR_LARGE_PKTLIMIT, so the final wire
* page in each direction is genuinely
* partial (42 entries), not just N full
* pages */
/* Shared page-size cap for both directions in this test -- == the REAL
* production XCECD_RESYNC_REPORT_PKTLIMIT for the report direction; the
* query direction has no real production constant (see header comment
* above / CONCERNS, bug-1755), so this test uses the same value for
* symmetry while directly exercising xce_job_state_query_req()'s already-
* implemented pagination loop. */
#define TWR_LARGE_PKTLIMIT 100
#define TWR_LARGE_N_PAGES \
((TWR_LARGE_N_TOTAL + TWR_LARGE_PKTLIMIT - 1) / TWR_LARGE_PKTLIMIT)
static uint64_t twr_large_jobid(int i)
{
return TWR_JOBID_LARGE_BASE + (uint64_t)i;
}
/* 0 = running, 1 = finished-success, 2 = finished-failure, 3 =
* unrecognized -- see this test's own header comment on why the group
* boundaries are deliberately not page-aligned. */
static int twr_large_group(int i)
{
if (i < TWR_LARGE_N_RUNNING)
return 0;
if (i < TWR_LARGE_N_RUNNING + TWR_LARGE_N_FINOK)
return 1;
if (i < TWR_LARGE_N_RUNNING + TWR_LARGE_N_FINOK + TWR_LARGE_N_FINFAIL)
return 2;
return 3;
}
static void test_large_job_count_pagination_correctness(void)
{
char tmp[256];
uint64_t finok_anchor_s;
strcpy(tmp, "/tmp/xce_twr74_large_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
/* Deliberately far from "now" -- same defensive idiom as every other
* finish-time test in this file. */
finok_anchor_s = (uint64_t)time(NULL) - 3600;
/* Phase 1: real cluster (workermgr + resmgr + jobmgr + hooksys) + real
* disk persist -- drives the REAL rejoin/query-construction chain over
* TWR_LARGE_N_TOTAL jobs, then a REAL, explicitly-bounded wire query
* send and a REAL, real-production-bounded wire report send, both
* genuinely multi-page. Own local cluster_t (not the shared g_twr
* fixture), mirroring every other restart-replay test in this file. */
{
cluster_t c;
kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 64, .delta = 16 };
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 1024, .delta = 256 };
persist_config_t cfg;
worker_t* w;
job_t** jobs;
hostattrs_t attrs, new_attrs;
xce_hostconfig_t hconfig;
endpoint_t stack_ep;
endpoint_t* ep;
xce_job_state_entry_t* storage;
xce_job_state_entry_t** entries;
xce_job_state_query_t query;
xce_job_state_report_t report;
int i, rc;
uint64_t lsn_before, lsn_after;
int calls_before, calls_after;
static const char* res_names[TWR_RT_N_RES] = {
"type", "model", "cpuf", "maxcpus", "maxmem",
"maxswap", "maxtmp", "ndisks", "slots"
};
static const int res_valtypes[TWR_RT_N_RES] = {
XCE_VAL_STRING, XCE_VAL_STRING,
XCE_VAL_NUMERIC, XCE_VAL_NUMERIC, XCE_VAL_NUMERIC,
XCE_VAL_NUMERIC, XCE_VAL_NUMERIC, XCE_VAL_NUMERIC,
XCE_VAL_NUMERIC
};
if (makeString((char*)"twr74-probe") == NULL)
xstring_table_init(64);
if (!g_twr_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_twr_jobobj_ready = 1;
}
memset(&c, 0, sizeof(c));
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&c.hooksys, XCE_CONTROLLER_HOOK_MAX));
c.workermgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(c.workermgr);
c.resmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(c.resmgr);
c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
TEST_ASSERT_NOT_NULL(c.jobmgr);
for (i = 0; i < TWR_RT_N_RES; i++) {
xce_resource_t rdef;
memset(&rdef, 0, sizeof(rdef));
rdef.name = (char*)res_names[i];
rdef.valtype = res_valtypes[i];
rdef.flags = 0;
rdef.idx = -1;
TEST_ASSERT_NOT_NULL(makeResource(c.resmgr, &rdef));
}
c.myhost_info.name = (char*)"twr74-controller";
c.myhost_info.key = (char*)"twr74-key";
c.myhost = &c.myhost_info;
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
persist_set_workdir(&c.persist, tmp, "ctrl-test");
TEST_ASSERT_EQUAL_INT(0, persist_init(&c.persist, &cfg, NULL));
TEST_ASSERT_EQUAL_INT(0, persist_start(&c.persist));
jobs = (job_t**)malloc(sizeof(job_t*) * TWR_LARGE_N_TOTAL);
TEST_ASSERT_NOT_NULL(jobs);
/* Seed a durable WAL_JOB_SUBMIT fact + a matching in-memory job_t
* for all TWR_LARGE_N_TOTAL jobs -- all RUN, all assigned to the
* SAME worker host, exactly what the controller believes right
* before this rejoin (bypassing dispatch/replay, mirroring every
* sibling restart-replay test in this file). */
for (i = 0; i < TWR_LARGE_N_TOTAL; i++) {
uint64_t jobid = twr_large_jobid(i);
job_submit_data_t wal_submit;
xce_job_t wal_jdef;
uint8_t* buf = NULL;
size_t sz = 0;
job_submit_data_t submit;
xce_job_t jdef;
memset(&wal_submit, 0, sizeof(wal_submit));
wal_submit.queue = (char*)"q";
wal_submit.user = (char*)"tester";
memset(&wal_jdef, 0, sizeof(wal_jdef));
wal_jdef.jobid = jobid;
wal_jdef.state = JOB_STAT_RUN;
wal_jdef.submit = &wal_submit;
TEST_ASSERT_EQUAL_INT(0, twr_encode_job(&wal_jdef, &buf, &sz));
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(&c.persist, WAL_JOB_SUBMIT, buf,
(uint32_t)sz));
free(buf);
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)"q";
submit.user = (char*)"tester";
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = JOB_STAT_RUN;
jdef.submit = &submit;
jobs[i] = makeJob(c.jobmgr, &jdef);
TEST_ASSERT_NOT_NULL(jobs[i]);
jobs[i]->job.allocs
= twr_make_test_alloc_for_host(jobid, "resync-large-host");
TEST_ASSERT_NOT_NULL(jobs[i]->job.allocs);
}
/* ── Join: rejoin_worker (the REAL join entry point) arms the round
* over all TWR_LARGE_N_TOTAL non-terminal jobs -- Req 1.1, 2.1, and
* the "no artificial cap" proof (Req 8.1) that resync_collect_
* query_jobids' realloc-doubling growth (resync.c) genuinely has no
* fixed-size ceiling. */
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;
{
xce_worker_t host;
memset(&host, 0, sizeof(host));
host.name = (char*)"resync-large-host";
host.attrs = &attrs;
w = makeWorker(c.workermgr, &host);
}
TEST_ASSERT_NOT_NULL(w);
memset(&new_attrs, 0, sizeof(new_attrs));
new_attrs.hosttype = makeString("LINUX");
new_attrs.hostmodel = makeString("x86");
new_attrs.cpufactor = 1.0f;
new_attrs.maxcpus = 4;
new_attrs.maxmem = 16;
new_attrs.maxswap = 32;
new_attrs.maxtmp = 64;
new_attrs.ndisks = 1;
memset(&hconfig, 0, sizeof(hconfig));
hconfig.hosttype = (char*)"!";
hconfig.hostmodel = (char*)"!";
hconfig.mxj = 4;
memset(&stack_ep, 0, sizeof(stack_ep));
rc = rejoin_worker(&c, w, &stack_ep, &new_attrs, &hconfig);
TEST_ASSERT_GREATER_OR_EQUAL(0, rc);
TEST_ASSERT_NOT_NULL_MESSAGE(w->resync,
"a rejoin with TWR_LARGE_N_TOTAL known non-terminal jobs on "
"this host must arm a reconciliation round (Req 1.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(TWR_LARGE_N_TOTAL,
w->resync->n_queried_jobids,
"the round's query set must match exactly the host's non-"
"terminal jobs, with no artificial cap on how many resync_"
"arm_round will collect (Req 1.1, 2.1, 8.1)");
TEST_ASSERT_TRUE(IS_WORKER_RECONCILING(w->host.status));
ep = twr_rt_new_test_endpoint();
/* ── QUERY direction: a REAL wire-encoded xce_job_state_query_req()
* send, called directly (not via worker_send_resync_query(), which
* hardcodes pktlimit=0 -- see this test's own header comment and
* CONCERNS) with an explicit page size, genuinely producing
* multiple wire pages for this job count. */
query = (xce_job_state_query_t){
.n_total = w->resync->n_queried_jobids,
.pos = 0,
.count = w->resync->n_queried_jobids,
.jobids = w->resync->queried_jobids
};
TEST_ASSERT_EQUAL_INT(0,
xce_job_state_query_req(ep, c.myhost, &query, NULL, NULL,
TWR_LARGE_PKTLIMIT));
{
bool seen_query[TWR_LARGE_N_TOTAL];
int running_total = 0;
int query_pages = 0;
memset(seen_query, 0, sizeof(seen_query));
while (running_total < TWR_LARGE_N_TOTAL) {
xce_request_t decoded_query;
int j;
twr_rt_pop_and_decode(ep, &decoded_query);
TEST_ASSERT_EQUAL_INT(OPCode_jobStateQuery,
decoded_query.opcode);
TEST_ASSERT_EQUAL_INT(Request_body_jobstatequerymsg,
decoded_query.body_which);
TEST_ASSERT_NOT_NULL(decoded_query.body.jobstatequerymsg);
TEST_ASSERT_EQUAL_INT_MESSAGE(TWR_LARGE_N_TOTAL,
decoded_query.body.jobstatequerymsg->n_total,
"Req 2.1: every query page must carry the SAME full "
"n_total, not a per-page total");
TEST_ASSERT_EQUAL_INT_MESSAGE(running_total,
decoded_query.body.jobstatequerymsg->pos,
"Req 8.3: each page's pos must pick up exactly where "
"the previous page's count left off -- no gap, no "
"overlap");
for (j = 0; j < decoded_query.body.jobstatequerymsg->count;
j++) {
uint64_t jobid
= decoded_query.body.jobstatequerymsg->jobids[j];
uint64_t idx = jobid - TWR_JOBID_LARGE_BASE;
TEST_ASSERT_TRUE_MESSAGE(
idx < (uint64_t)TWR_LARGE_N_TOTAL,
"a jobid appearing on the wire query must be one "
"of the armed round's own jobids");
TEST_ASSERT_FALSE_MESSAGE(seen_query[idx],
"Req 8.3: no jobid may be sent twice across query "
"pages");
seen_query[idx] = true;
}
running_total
+= decoded_query.body.jobstatequerymsg->count;
query_pages++;
xce_free_request(&decoded_query);
}
TEST_ASSERT_EQUAL_INT_MESSAGE(TWR_LARGE_N_PAGES, query_pages,
"Req 8.1-8.3: an order-of-magnitude-larger-than-one-page "
"query must genuinely span many wire pages, including a "
"partial final page, not silently collapse to one (the "
"real production worker_send_resync_query() call site "
"would collapse to one -- see CONCERNS, bug-1755 -- this "
"assertion is against xce_job_state_query_req()'s own "
"pagination mechanism, exercised directly)");
for (i = 0; i < TWR_LARGE_N_TOTAL; i++) {
TEST_ASSERT_TRUE_MESSAGE(seen_query[i],
"Req 8.3: every armed jobid must appear exactly once "
"across all query pages -- none silently dropped past "
"a page boundary");
}
}
/* ── Worker's answer: a REAL wire-encoded-then-decoded OPCode_
* jobStateReport, sent at the REAL production report page size
* (XCECD_RESYNC_REPORT_PKTLIMIT, controller_connector.c) --
* genuinely many wire pages -- covering all four outcome kinds,
* fed page-by-page into resync_handle_report_page() (task 6.2's
* real reassembly entry point), never resync_apply_report()
* directly. */
storage = (xce_job_state_entry_t*)malloc(
sizeof(xce_job_state_entry_t) * TWR_LARGE_N_TOTAL);
entries = (xce_job_state_entry_t**)malloc(
sizeof(xce_job_state_entry_t*) * TWR_LARGE_N_TOTAL);
TEST_ASSERT_NOT_NULL(storage);
TEST_ASSERT_NOT_NULL(entries);
for (i = 0; i < TWR_LARGE_N_TOTAL; i++) {
xce_job_state_entry_t* e = &storage[i];
memset(e, 0, sizeof(*e));
e->jobid = twr_large_jobid(i);
e->queried = 1;
switch (twr_large_group(i)) {
case 0:
e->status = JobStateStatus_running;
break;
case 1: {
int local = i - TWR_LARGE_N_RUNNING;
uint64_t fin_s = finok_anchor_s - (uint64_t)(local * 3);
e->status = JobStateStatus_finishedOk;
e->finish_time_ms = fin_s * 1000ULL;
break;
}
case 2: {
int local = i - TWR_LARGE_N_RUNNING - TWR_LARGE_N_FINOK;
e->status = JobStateStatus_finishedFail;
e->exit_code = 200 + local;
break;
}
default:
e->status = JobStateStatus_unrecognized;
break;
}
entries[i] = e;
}
report = (xce_job_state_report_t){
.n_total = TWR_LARGE_N_TOTAL,
.pos = 0,
.count = TWR_LARGE_N_TOTAL,
.entries = entries
};
TEST_ASSERT_EQUAL_INT(0,
xce_job_state_report_req(ep, c.myhost, &report, NULL, NULL,
TWR_LARGE_PKTLIMIT));
g_twr_update_hook_fires = 0;
TEST_ASSERT_EQUAL_INT(0,
xce_add_hook(&c.hooksys, XCE_CONTROLLER_HOOK_JOB,
twr_job_update_spy_hook, NULL));
pthread_mutex_lock(&c.persist.lock);
lsn_before = c.persist.current_lsn;
pthread_mutex_unlock(&c.persist.lock);
resync_apply_report_call_count_reset_for_test();
calls_before = resync_apply_report_call_count_for_test();
{
int running_total = 0;
int report_pages = 0;
while (running_total < TWR_LARGE_N_TOTAL) {
xce_request_t decoded_report;
int rc_page;
twr_rt_pop_and_decode(ep, &decoded_report);
TEST_ASSERT_EQUAL_INT(OPCode_jobStateReport,
decoded_report.opcode);
TEST_ASSERT_EQUAL_INT(Request_body_jobstatereportmsg,
decoded_report.body_which);
TEST_ASSERT_NOT_NULL(decoded_report.body.jobstatereportmsg);
running_total
+= decoded_report.body.jobstatereportmsg->count;
report_pages++;
rc_page = resync_handle_report_page(
&c, w, decoded_report.body.jobstatereportmsg);
if (running_total < TWR_LARGE_N_TOTAL) {
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc_page,
"a non-final report page must not complete the "
"report");
} else {
TEST_ASSERT_EQUAL_INT_MESSAGE(1, rc_page,
"the final (partial) report page must complete "
"the report");
}
xce_free_request(&decoded_report);
}
TEST_ASSERT_EQUAL_INT_MESSAGE(TWR_LARGE_N_PAGES, report_pages,
"Req 8.1-8.3: at the real production per-page limit, this "
"job count must genuinely span many wire pages, including "
"a partial final page");
}
calls_after = resync_apply_report_call_count_for_test();
TEST_ASSERT_EQUAL_INT_MESSAGE(1, calls_after - calls_before,
"resync_apply_report must be invoked EXACTLY ONCE for the "
"whole large, many-page report -- not once per page (task "
"6.2's own structural proof, generalized from 2 pages to "
"double digits)");
pthread_mutex_lock(&c.persist.lock);
lsn_after = c.persist.current_lsn;
pthread_mutex_unlock(&c.persist.lock);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
lsn_before
+ (uint64_t)(TWR_LARGE_N_FINOK + TWR_LARGE_N_FINFAIL
+ TWR_LARGE_N_UNREC),
lsn_after,
"EXACTLY one durable fact per non-running job (finished-"
"success x WAL_JOB_FINISH + finished-failure x WAL_JOB_FAIL + "
"unrecognized x WAL_JOB_UPDATE), and NONE for the still-"
"running jobs -- an exact count (not merely a lower bound) is "
"the genuine 'none dropped, none duplicated across page "
"boundaries' proof");
TEST_ASSERT_EQUAL_INT_MESSAGE(TWR_LARGE_N_UNREC,
g_twr_update_hook_fires,
"Req 7.1-7.3: the EAGER unrecognized-to-pending transition "
"must reach the scheduler's/apiserver's notify path "
"synchronously, exactly once per unrecognized job, regardless "
"of which page it happened to land on");
TEST_ASSERT_NULL_MESSAGE(w->resync,
"once every queried job across every report page has a "
"reconciled outcome, the round must resolve (Req 5.1, 5.2)");
TEST_ASSERT_FALSE_MESSAGE(IS_WORKER_RECONCILING(w->host.status),
"XCE_WORKER_RECONCILING must clear once the round resolves");
/* Phase 1 end-to-end per-job assertions (Req 8.1-8.3: EVERY job
* accounted for, not a sample) -- everything observable before any
* commit round-trip. */
for (i = 0; i < TWR_LARGE_N_TOTAL; i++) {
switch (twr_large_group(i)) {
case 0:
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN,
jobs[i]->job.state,
"still-running (queried) must be left exactly as-is");
TEST_ASSERT_NOT_NULL(jobs[i]->job.allocs);
break;
case 1:
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN,
jobs[i]->job.state,
"finished-success stays DEFERRED (record_job_finish_"
"fact_at) -- mutates later via apply_committed_entry, "
"not synchronously here");
break;
case 2:
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN,
jobs[i]->job.state,
"finished-failure stays DEFERRED (on_job_fail/record_"
"job_fail_fact), same as finished-success");
break;
default:
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND,
jobs[i]->job.state,
"unrecognized reconciles EAGERLY -- PEND "
"synchronously, before any commit round-trip");
TEST_ASSERT_NULL_MESSAGE(jobs[i]->job.allocs,
"bug-1631: reconciled-to-pending must release the "
"allocation synchronously, at scale, across page "
"boundaries");
break;
}
}
xce_disable_hook(&c.hooksys, XCE_CONTROLLER_HOOK_JOB,
twr_job_update_spy_hook);
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c.persist));
persist_stop(&c.persist);
persist_fini(&c.persist);
if (c.workermgr) dereference(c.workermgr);
if (c.resmgr) dereference(c.resmgr);
if (c.jobmgr) dereference(c.jobmgr);
xce_hook_fini(&c.hooksys);
dereference(new_attrs.hosttype);
dereference(new_attrs.hostmodel);
endpoint_fini(ep);
free(ep);
free(jobs);
free(storage);
free(entries);
}
/* Phase 2: fresh persist_t + fresh cluster_t (jobmgr/hooksys only),
* recover from disk -- simulates a controller restart, settling the two
* DEFERRED outcomes (finished-success/failure) across every page they
* happened to land on, and re-confirming the EAGER unrecognized->
* pending outcome also round-trips through a real WAL replay, not just
* the live eager-mutation path -- for EVERY one of the TWR_LARGE_N_
* TOTAL jobs, individually, not a sample. */
{
persist_t* p2 = twr_make_fresh_persist(tmp);
cluster_t c2;
job_t** jobs2;
int rc, i;
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 1024, .delta = 256 };
TEST_ASSERT_NOT_NULL(p2);
memset(&c2, 0, sizeof(c2));
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&c2.hooksys, XCE_CONTROLLER_HOOK_MAX));
c2.jobmgr = makeInstance(&xkvmgr_type, &int_params);
TEST_ASSERT_NOT_NULL(c2.jobmgr);
g_twr_update_hook_fires = 0;
TEST_ASSERT_EQUAL_INT(0,
xce_add_hook(&c2.hooksys, XCE_CONTROLLER_HOOK_JOB,
twr_job_update_spy_hook, NULL));
p2->cluster = &c2;
rc = persist_recover(p2);
TEST_ASSERT_EQUAL_INT(0, rc);
jobs2 = (job_t**)malloc(sizeof(job_t*) * TWR_LARGE_N_TOTAL);
TEST_ASSERT_NOT_NULL(jobs2);
lock(c2.jobmgr, RTTI_LOCK_READ);
for (i = 0; i < TWR_LARGE_N_TOTAL; i++) {
jobs2[i] = NULL;
key_find(c2.jobmgr, (void*)twr_large_jobid(i),
(void**)&jobs2[i]);
}
unlock(c2.jobmgr);
for (i = 0; i < TWR_LARGE_N_TOTAL; i++) {
TEST_ASSERT_NOT_NULL_MESSAGE(jobs2[i],
"restart replay must reconstruct every job via its own "
"WAL_JOB_SUBMIT fact");
switch (twr_large_group(i)) {
case 0:
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN,
jobs2[i]->job.state,
"still-running: no fact was ever recorded, so replay "
"leaves it exactly as WAL_JOB_SUBMIT alone "
"reconstructs it");
break;
case 1: {
int local = i - TWR_LARGE_N_RUNNING;
uint64_t fin_s = finok_anchor_s - (uint64_t)(local * 3);
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE,
jobs2[i]->job.state,
"finished-success: the DEFERRED record_job_finish_"
"fact_at fact must apply to DONE on replay");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(fin_s,
(uint64_t)jobs2[i]->job.finish_time,
"each finished-success job must recover ITS OWN "
"worker-reported finish_time -- no cross-job mixing "
"across a page boundary");
break;
}
case 2:
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_EXIT,
jobs2[i]->job.state,
"finished-failure: the DEFERRED on_job_fail/record_"
"job_fail_fact fact must apply to EXIT on replay");
break;
default:
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND,
jobs2[i]->job.state,
"unrecognized: the EAGER record_job_reconcile_"
"pending_fact's durable WAL_JOB_UPDATE fact must ALSO "
"reconstruct correctly on replay");
TEST_ASSERT_NULL(jobs2[i]->job.allocs);
break;
}
}
TEST_ASSERT_EQUAL_INT_MESSAGE(
TWR_LARGE_N_FINOK + TWR_LARGE_N_FINFAIL + TWR_LARGE_N_UNREC,
g_twr_update_hook_fires,
"Req 7.1-7.3: replaying the durable facts this reconciliation "
"round produced must reach the scheduler's/apiserver's notify "
"path for EVERY actual state change -- finished-success, "
"finished-failure, and unrecognized-to-pending -- exactly "
"once each, across every job (still-running recorded no fact, "
"and WAL_JOB_SUBMIT fires HOOK_DETAIL_JOB_NEW, not JOB_UPDATE, "
"so neither adds to this count)");
xce_disable_hook(&c2.hooksys, XCE_CONTROLLER_HOOK_JOB,
twr_job_update_spy_hook);
xce_hook_fini(&c2.hooksys);
if (c2.jobmgr) dereference(c2.jobmgr);
twr_teardown_persist(p2);
free(jobs2);
}
{
char cmd[512];
snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp);
if (system(cmd) != 0)
fprintf(stderr, "cleanup warning: rm -rf %s failed\n", tmp);
}
printf("test_large_job_count_pagination_correctness: PASS\n");
}
/* ── Test group entry point ───────────────────────────────────────────── */
void test_worker_resync(void)
@ -4098,4 +4776,5 @@ void test_worker_resync(void)
RUN_TEST(test_full_rejoin_reconciliation_round_trip);
RUN_TEST(test_failover_during_job_recovers_lost_finish);
RUN_TEST(test_burst_recovery_unbounded_by_job_count);
RUN_TEST(test_large_job_count_pagination_correctness);
}