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

test(worker-rejoin-resync): task 7.3 - burst recovery unbounded by count

Browse source 
Add test_burst_recovery_unbounded_by_job_count, reproducing bug-1741's
exact original shape (7 jobs on one host, 6 lost finish reports) in
one reconciliation round and one wire page: 5 finished-success + 1
finished-failure recover correctly and distinctly, while an already-
DONE job re-reported as an unqueried extra (with a deliberately wrong
finish time) is proven structurally unchanged - exact +6 WAL LSN
delta, not >=, so a duplicate append is genuinely caught rather than
silently tolerated. Test-only, no production changes; deliberately
kept to a single page since page-boundary correctness is task 7.4's
separate concern.

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:04:37 +00:00
parent bceca0c990
commit 165fc1586a
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
2 changed files with 634 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -3436,6 +3436,637 @@ static void test_failover_during_job_recovers_lost_finish(void)
printf("test_failover_during_job_recovers_lost_finish: PASS\n");
}
/* ── Task 7.3: Burst/array recovery unbounded by job count -- bug-1741's OWN
* exact originally-reported shape (requirements.md's Introduction: "a burst
* of near-simultaneous job completions on one worker host lost 6 of 7
* finish reports, leaving those jobs stuck in RUN forever while their
* siblings all completed normally") ────────────────────────────────────────
*
* Task 7.1 proved the general four-outcome round trip (one job per
* outcome); task 7.2 proved the single-job lost-finish recovery in
* isolation. Neither proves what THIS task is for: bug-1741's own reported
* shape was SEVERAL jobs at once, MOST lost, with recovery required to
* scale correctly across all of them together in ONE round -- not merely
* repeat the single-job case N times with no cross-job interaction check
* (tasks.md's own Observable Completion Criterion for this task).
*
* 7 jobs total on ONE worker host (bug-1741's own literal numbers):
* - 1 job (TWR_JOBID_BURST_ALREADY_DONE) already finished successfully
* BEFORE this rejoin -- a durable WAL_JOB_SUBMIT + WAL_JOB_FINISH pair
* is already on disk (the FINISH appended directly via persist_submit_
* job_finish_fact(), the same primitive record_job_finish_fact_at()
* itself calls -- modeling a finish report that DID reach and was
* durably committed by an earlier controller instance, before this
* rejoin), in-memory state already DONE. Being terminal, it must NOT be
* part of the round's query set (Req 1.1, 1.2, 2.1) -- but the worker's
* own retention window (Req 3.6, 3.7) would still recognize it as
* "recently finished" and report it AGAIN, as an unqueried extra, this
* time carrying a deliberately WRONG (bogus, recent) finish_time_ms.
* This is this task's own "none duplicated" Observable Completion
* Criterion, made concrete: this job's already-recorded outcome must
* survive completely unchanged -- both synchronously (Phase 1) and
* after a real restart replay (Phase 2) -- never overwritten by the
* bogus re-report, and with NO new WAL fact appended for it at all.
* - 6 jobs (TWR_JOBID_BURST_LOST_1..6) mirror bug-1741's own "6 of 7 lost"
* ratio exactly: each is durably recorded RUN with NO finish fact of
* any kind ever appended (the lost fire-and-forget XCECLI_FF_CALL, the
* same starting condition as task 7.2's single-job case) -- all 6 are
* non-terminal and assigned to the SAME host, so all 6 are armed into
* ONE round's query set together (Req 2.1, 8.1). Each carries its OWN
* distinct worker-reported finish_time_ms/exit_code (5 finished-success
* + 1 finished-failure, a genuine mixed-outcome burst) -- distinct per
* job specifically so a bug that mixes up entries across the loop (an
* off-by-one / shared-variable defect in resync_apply_report's
* per-entry routing) would surface as a WRONG value under a SPECIFIC
* jobid, never masked by every job sharing one identical expected
* outcome.
*
* Drives the SAME real rejoin -> query-send -> report-apply chain tasks
* 7.1/7.2 already established (rejoin_worker, worker_send_resync_query's
* real wire OPCode_jobStateQuery, a real worker answer via xce_job_state_
* report_req's real wire OPCode_jobStateReport, resync_handle_report_page,
* never resync_apply_report directly) -- one single-page report covering
* all 7 entries together (6 queried + 1 unqueried extra). Deliberately NOT
* forced across multiple wire pages: task 7.4's own separately-scoped job
* is proving PAGINATION boundary correctness at a much larger count; this
* task's job is proving RECOVERY LOGIC correctness at bug-1741's own
* "several jobs at once" scale, per tasks.md's explicit boundary note
* separating the two. */
#define TWR_JOBID_BURST_ALREADY_DONE 81501ULL
#define TWR_JOBID_BURST_LOST_1 81502ULL
#define TWR_JOBID_BURST_LOST_2 81503ULL
#define TWR_JOBID_BURST_LOST_3 81504ULL
#define TWR_JOBID_BURST_LOST_4 81505ULL
#define TWR_JOBID_BURST_LOST_5 81506ULL
#define TWR_JOBID_BURST_LOST_6 81507ULL /* finished-FAILURE, not OK */
#define TWR_BURST_N_LOST 6
#define TWR_BURST_N_TOTAL 7 /* 1 already-done + 6 lost -- bug-1741's own 7/6 numbers */
static void test_burst_recovery_unbounded_by_job_count(void)
{
char tmp[256];
uint64_t already_done_finish_time_s;
uint64_t bogus_reapply_finish_time_s;
uint64_t lost_finish_time_s[TWR_BURST_N_LOST - 1]; /* LOST_1..LOST_5 (finished-ok) */
int32_t lost_fail_exit_code = 66;
strcpy(tmp, "/tmp/xce_twr73_burst_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
/* Deliberately distinct/far-from-"now" values throughout -- so a
* reconciliation-processing-time bug (time(NULL)) or a cross-job value
* mix-up could never accidentally produce a matching value, same
* defensive idiom this file's other finish-time tests already
* establish. */
already_done_finish_time_s = (uint64_t)time(NULL) - 7200;
bogus_reapply_finish_time_s = (uint64_t)time(NULL) - 30;
{
int k;
for (k = 0; k < TWR_BURST_N_LOST - 1; k++)
lost_finish_time_s[k]
= (uint64_t)time(NULL) - 3600 - (uint64_t)(k * 137);
}
/* Phase 1: real cluster (workermgr + resmgr + jobmgr + hooksys) + real
* disk persist -- drives the REAL join/query-send/report-apply chain
* over all 7 jobs together. 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 = 32, .delta = 16 };
persist_config_t cfg;
worker_t* w;
job_t* j_done;
job_t* j_lost[TWR_BURST_N_LOST];
uint64_t lost_ids[TWR_BURST_N_LOST] = {
TWR_JOBID_BURST_LOST_1, TWR_JOBID_BURST_LOST_2,
TWR_JOBID_BURST_LOST_3, TWR_JOBID_BURST_LOST_4,
TWR_JOBID_BURST_LOST_5, TWR_JOBID_BURST_LOST_6
};
hostattrs_t attrs, new_attrs;
xce_hostconfig_t hconfig;
endpoint_t stack_ep;
endpoint_t* ep;
xce_request_t decoded_query, decoded_report;
int saw[TWR_BURST_N_LOST];
int i, rc;
uint64_t lsn_before, lsn_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*)"twr73-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*)"twr73-controller";
c.myhost_info.key = (char*)"twr73-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));
/* Seed the ALREADY-DONE job: durable WAL_JOB_SUBMIT + a REAL,
* already-committed WAL_JOB_FINISH (appended directly via
* persist_submit_job_finish_fact -- modeling a finish report that
* reached and was durably recorded by an EARLIER controller
* instance, before this rejoin) -- plus a matching in-memory job_t
* already DONE, allocs already released, exactly what a genuinely
* finished job looks like. */
{
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 = TWR_JOBID_BURST_ALREADY_DONE;
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);
TEST_ASSERT_EQUAL_INT(0,
persist_submit_job_finish_fact(&c.persist,
TWR_JOBID_BURST_ALREADY_DONE, JOB_STAT_DONE,
already_done_finish_time_s, 0));
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)"q";
submit.user = (char*)"tester";
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = TWR_JOBID_BURST_ALREADY_DONE;
jdef.state = JOB_STAT_DONE;
jdef.submit = &submit;
j_done = makeJob(c.jobmgr, &jdef);
TEST_ASSERT_NOT_NULL(j_done);
j_done->job.finish_time = (time_t)already_done_finish_time_s;
j_done->job.exit_info
= (job_exitinfo_t*)calloc(1, sizeof(job_exitinfo_t));
TEST_ASSERT_NOT_NULL(j_done->job.exit_info);
j_done->job.exit_info->exitcode = 0;
TEST_ASSERT_NULL(j_done->job.allocs);
}
/* Seed the 6 LOST jobs: durable WAL_JOB_SUBMIT only, RUN, all
* assigned to the SAME host -- NO finish fact of any kind, the
* lost-fire-and-forget-send starting condition, x6 (bug-1741's own
* "6 of 7 lost" ratio). */
for (i = 0; i < TWR_BURST_N_LOST; 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 = lost_ids[i];
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 = lost_ids[i];
jdef.state = JOB_STAT_RUN;
jdef.submit = &submit;
j_lost[i] = makeJob(c.jobmgr, &jdef);
TEST_ASSERT_NOT_NULL(j_lost[i]);
j_lost[i]->job.allocs = twr_make_test_alloc_for_host(
lost_ids[i], "resync-burst-host");
TEST_ASSERT_NOT_NULL(j_lost[i]->job.allocs);
}
/* ── Join: rejoin_worker (the REAL join entry point) arms the round
* over exactly the 6 non-terminal (lost) jobs -- the 7th,
* already-DONE job must NOT be part of the query set (Req 1.1,
* 1.2, 2.1). */
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-burst-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 6 known non-terminal jobs on this host must arm "
"a reconciliation round (Req 1.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(TWR_BURST_N_LOST,
w->resync->n_queried_jobids,
"the round's query set must match exactly the host's 6 "
"non-terminal jobs -- the already-DONE 7th job must be "
"excluded (Req 1.1, 1.2, 2.1)");
TEST_ASSERT_TRUE(IS_WORKER_RECONCILING(w->host.status));
ep = twr_rt_new_test_endpoint();
TEST_ASSERT_EQUAL_INT(0, worker_send_resync_query(&c, w, ep));
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_BURST_N_LOST,
decoded_query.body.jobstatequerymsg->n_total,
"the REAL wire query (Req 2.1, 2.2, 8.1) must list exactly the "
"6 lost jobs, regardless of the burst also containing a 7th, "
"already-terminal one");
for (i = 0; i < TWR_BURST_N_LOST; i++)
saw[i] = 0;
for (i = 0; i < decoded_query.body.jobstatequerymsg->count; i++) {
uint64_t id = decoded_query.body.jobstatequerymsg->jobids[i];
int k;
TEST_ASSERT_FALSE_MESSAGE(id == TWR_JOBID_BURST_ALREADY_DONE,
"the already-terminal job must never appear in the query "
"set (Req 1.2)");
for (k = 0; k < TWR_BURST_N_LOST; k++) {
if (id == lost_ids[k])
saw[k] = 1;
}
}
for (i = 0; i < TWR_BURST_N_LOST; i++) {
TEST_ASSERT_TRUE_MESSAGE(saw[i],
"every one of the 6 lost jobs must be covered by the wire "
"query, by unique ID (Req 2.2, 8.1)");
}
xce_free_request(&decoded_query);
/* ── Worker's real answer: all 6 lost jobs report their TRUE
* outcome (5 finished-success + 1 finished-failure, each with its
* OWN distinct finish_time_ms/exit_code -- Req 3.1, 3.4, 3.5), plus
* the 7th, already-DONE job reported AGAIN as an unqueried extra
* (Req 3.6) -- carrying a deliberately WRONG finish_time_ms, to
* prove reconciliation does not clobber/duplicate what was already
* correctly recorded (this task's own "none duplicated" Observable
* Completion Criterion). All 7 entries travel in ONE page (Req
* 8.1, 8.2) -- this task is not about page-boundary mechanics
* (task 7.4's own separate job). */
{
xce_job_state_entry_t e_lost[TWR_BURST_N_LOST];
xce_job_state_entry_t e_already_done;
xce_job_state_entry_t* entries[TWR_BURST_N_TOTAL];
xce_job_state_report_t report;
for (i = 0; i < TWR_BURST_N_LOST - 1; i++) {
e_lost[i] = (xce_job_state_entry_t){
.jobid = lost_ids[i],
.status = JobStateStatus_finishedOk,
.exit_code = 0,
.finish_time_ms = lost_finish_time_s[i] * 1000ULL,
.queried = 1
};
}
/* The 6th lost job: finished-FAILURE, not success -- a genuine
* mixed-outcome burst (this task's own scope), not a re-proof
* of 7.1's already-established per-outcome routing in
* isolation. */
e_lost[TWR_BURST_N_LOST - 1] = (xce_job_state_entry_t){
.jobid = lost_ids[TWR_BURST_N_LOST - 1],
.status = JobStateStatus_finishedFail,
.exit_code = lost_fail_exit_code,
.finish_time_ms = 0,
.queried = 1
};
e_already_done = (xce_job_state_entry_t){
.jobid = TWR_JOBID_BURST_ALREADY_DONE,
.status = JobStateStatus_finishedOk,
.exit_code = 0,
.finish_time_ms = bogus_reapply_finish_time_s * 1000ULL,
.queried = 0
};
for (i = 0; i < TWR_BURST_N_LOST; i++)
entries[i] = &e_lost[i];
entries[TWR_BURST_N_LOST] = &e_already_done;
report = (xce_job_state_report_t){
.n_total = TWR_BURST_N_TOTAL, .pos = 0,
.count = TWR_BURST_N_TOTAL, .entries = entries
};
TEST_ASSERT_EQUAL_INT(0,
xce_job_state_report_req(ep, c.myhost, &report, NULL, NULL,
0));
}
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);
TEST_ASSERT_EQUAL_INT_MESSAGE(TWR_BURST_N_TOTAL,
decoded_report.body.jobstatereportmsg->n_total,
"the real wire report must carry all 7 entries in one page "
"(Req 8.1, 8.2)");
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);
rc = resync_handle_report_page(&c, w,
decoded_report.body.jobstatereportmsg);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, rc,
"a single-page report (count == n_total) must complete the "
"report immediately, regardless of the burst's job count "
"(Req 8.1)");
pthread_mutex_lock(&c.persist.lock);
lsn_after = c.persist.current_lsn;
pthread_mutex_unlock(&c.persist.lock);
xce_free_request(&decoded_report);
/* ── Phase 1 assertions: everything observable before any commit
* round trip. */
for (i = 0; i < TWR_BURST_N_LOST; i++) {
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN,
j_lost[i]->job.state,
"finished-success/failure both stay on the DEFERRED "
"record_job_finish_fact_at/record_job_fail_fact pattern -- "
"state mutates later via apply_committed_entry (Phase 2), "
"not synchronously here, for every one of the 6 lost "
"jobs");
}
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j_done->job.state,
"the already-DONE job must be left completely untouched by "
"its own bogus re-report (Req 4.3, 4.4 -- 'none duplicated')");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(already_done_finish_time_s,
(uint64_t)j_done->job.finish_time,
"the already-DONE job's ORIGINAL finish_time must survive "
"unchanged -- NOT overwritten by the bogus finish_time_ms "
"carried in its own unqueried re-report");
TEST_ASSERT_NOT_NULL(j_done->job.exit_info);
TEST_ASSERT_EQUAL_INT(0, j_done->job.exit_info->exitcode);
TEST_ASSERT_NULL(j_done->job.allocs);
TEST_ASSERT_NULL_MESSAGE(w->resync,
"once every one of the 6 queried jobs has a reconciled "
"outcome, the round must resolve regardless of the burst's "
"count (Req 5.1, 5.2, 8.1)");
TEST_ASSERT_FALSE_MESSAGE(IS_WORKER_RECONCILING(w->host.status),
"XCE_WORKER_RECONCILING must clear once the round resolves");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_twr_update_hook_fires,
"no EAGER outcome is present in this burst (no unrecognized, "
"no genuinely-unassigned extra) -- the 6 deferred finished-"
"success/failure facts have not applied yet, and the "
"already-DONE job's re-report is correctly a pure no-op with "
"no notify at all (Req 4.3, 4.4)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(
lsn_before + (uint64_t)TWR_BURST_N_LOST, lsn_after,
"EXACTLY one durable fact per lost job (5x WAL_JOB_FINISH + 1x "
"WAL_JOB_FAIL) must be appended for this report -- the "
"already-DONE job's duplicate re-report must append NONE, so "
"an exact count (not merely a lower bound) is the genuine "
"'none duplicated' proof: a regression that re-appended a "
"fact for the already-DONE job would silently pass a "
"'>=' check but fails this exact one");
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);
}
/* Phase 2: fresh persist_t + fresh cluster_t (jobmgr/hooksys only),
* recover from disk -- simulates a controller restart, the SAME apply_
* committed_entry() seam a live controller's own commit-apply pump
* would invoke to settle the 6 DEFERRED outcomes above. This is the
* test's core, observable completion criterion: every one of the 6
* lost jobs reaches its correct final state, individually and by its
* OWN distinct worker-reported value (not some other job's from the
* same burst), and the 7th job's already-correct outcome is confirmed
* to have survived completely untouched -- through a real restart
* replay, not just the live path. */
{
persist_t* p2 = twr_make_fresh_persist(tmp);
cluster_t c2;
job_t* j2_done = NULL;
job_t* j2_lost[TWR_BURST_N_LOST] = { NULL };
uint64_t lost_ids[TWR_BURST_N_LOST] = {
TWR_JOBID_BURST_LOST_1, TWR_JOBID_BURST_LOST_2,
TWR_JOBID_BURST_LOST_3, TWR_JOBID_BURST_LOST_4,
TWR_JOBID_BURST_LOST_5, TWR_JOBID_BURST_LOST_6
};
int rc, i;
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32, .delta = 16 };
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);
lock(c2.jobmgr, RTTI_LOCK_READ);
key_find(c2.jobmgr, (void*)TWR_JOBID_BURST_ALREADY_DONE,
(void**)&j2_done);
for (i = 0; i < TWR_BURST_N_LOST; i++)
key_find(c2.jobmgr, (void*)lost_ids[i], (void**)&j2_lost[i]);
unlock(c2.jobmgr);
TEST_ASSERT_NOT_NULL(j2_done);
for (i = 0; i < TWR_BURST_N_LOST; i++) {
TEST_ASSERT_NOT_NULL_MESSAGE(j2_lost[i],
"every one of the 6 lost jobs must reconstruct from its "
"own WAL_JOB_SUBMIT fact");
}
/* The core completion criterion (bug-1741, 'none left stuck'):
* every one of the 6 lost jobs, individually, by its own jobid,
* reaches its correct final state AND its OWN distinct
* worker-reported value -- proving no cross-job mixing across the
* burst (Req 8.1). */
for (i = 0; i < TWR_BURST_N_LOST - 1; i++) {
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE,
j2_lost[i]->job.state,
"each lost finished-success job must recover to DONE, "
"regardless of the burst's count (Req 8.1)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lost_finish_time_s[i],
(uint64_t)j2_lost[i]->job.finish_time,
"each lost job must recover ITS OWN worker-reported "
"finish_time -- not another job's from the same burst "
"(cross-job mixing regression check, Req 4.2, 3.5)");
}
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_EXIT,
j2_lost[TWR_BURST_N_LOST - 1]->job.state,
"the one finished-FAILURE job in the burst must recover to "
"EXIT, not DONE, distinctly from its 5 finished-success "
"siblings in the SAME round");
/* Req 4.2 remains documented-unsatisfiable for FAIL's own
* finish_time specifically (bug-1747, pre-existing, systemic, out
* of this task's boundary -- see tasks 7.1/7.2's own identical
* finding); exit_info stays NULL for the same pre-existing reason,
* asserted honestly rather than as a wrong "correct" value this
* codebase cannot actually produce. */
TEST_ASSERT_NULL(j2_lost[TWR_BURST_N_LOST - 1]->job.exit_info);
/* The 'none duplicated' completion criterion, confirmed to
* survive a restart replay too: the already-DONE job's ORIGINAL
* outcome, not the bogus re-reported one. */
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j2_done->job.state,
"the already-DONE job must still be DONE after replay");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(already_done_finish_time_s,
(uint64_t)j2_done->job.finish_time,
"the already-DONE job's finish_time after replay must still "
"be its ORIGINAL value -- the bogus re-report's "
"finish_time_ms must never have been durably recorded at "
"all");
TEST_ASSERT_EQUAL_INT_MESSAGE(TWR_BURST_N_TOTAL,
g_twr_update_hook_fires,
"replaying the durable facts in this log must reach the "
"scheduler's/apiserver's notify path exactly once per REAL "
"state change -- 6 for the 6 lost jobs' newly-recorded facts, "
"PLUS 1 for the already-DONE job's own ORIGINALLY-seeded "
"WAL_JOB_FINISH fact (recorded before this round even ran, "
"modeling an earlier controller instance's already-committed "
"finish): this bare-persist fixture runs no live apply pump, "
"so THIS replay is the first time apply_committed_entry ever "
"applies that pre-existing fact too, exactly like every other "
"seeded WAL_JOB_SUBMIT in this test -- NOT a second, "
"duplicate application of it. The genuine 'none duplicated' "
"proof is not this count (which is 7 either way) but that the "
"already-DONE job's finish_time below is still its ORIGINAL "
"value, and that Phase 1's own exact LSN-delta check already "
"proved no fact was durably appended for its bogus re-report "
"(Req 7.1-7.3)");
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);
}
{
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_burst_recovery_unbounded_by_job_count: PASS\n");
}
/* ── Test group entry point ───────────────────────────────────────────── */
void test_worker_resync(void)
@ -3466,4 +4097,5 @@ void test_worker_resync(void)
RUN_TEST(test_handle_report_page_malformed_bounds_rejected_and_recovers);
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);
}