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

test(worker-rejoin-resync): task 7.2 - failover-during-job recovery

Browse source 
Add test_failover_during_job_recovers_lost_finish, a focused proof of
the feature's own original motivating scenario (bug-1741): a job
dispatched and RUN with no finish fact ever committed anywhere
(modeling a lost fire-and-forget finish notify), recovered to DONE
with the worker-reported finish time - not stuck in RUN - via the
real rejoin/query/report/apply chain. Test-only, no production
changes; reuses task 7.1's helper infrastructure rather than
duplicating its broader four-outcome coverage.

Updates the existing bug-924 (/tmp exhaustion / flaky teardown) entry
with a 5th occurrence hit during validation, unrelated to this task's
own change.

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 23:19:15 +00:00
parent 8650cfd9ba
commit bceca0c990
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
3 changed files with 438 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -3007,6 +3007,435 @@ static void test_full_rejoin_reconciliation_round_trip(void)
printf("test_full_rejoin_reconciliation_round_trip: PASS\n");
}
/* ── Task 7.2: Failover-during-job recovery -- the feature's own core
* motivating scenario (bug-1741; requirements.md's Introduction; design.md
* "Testing Strategy" > "Integration Tests": "Failover-during-job: a job's
* finish notify is lost across a primary failover; the new primary's
* rejoin reconciliation recovers it (this is the feature's core bug-1741
* regression scenario)") ────────────────────────────────────────────────
*
* Proves the SIMPLEST version of bug-1741's own reported failure: one job,
* one lost finish report, one failover, one rejoin. bug-1741 itself was a
* burst of 7 near-simultaneous finish reports losing 6 of them; that
* burst/many-jobs shape is Requirement 8's own separate coverage (already
* proven by the multi-page tests above and by test_full_rejoin_
* reconciliation_round_trip's four-outcome round trip) -- this test's own
* job is only to name and prove the single-job case in isolation, not to
* re-prove the general round-trip machinery task 7.1 already covers.
*
* Per requirements.md's Introduction, today a worker's finish report is a
* single fire-and-forget XCECLI_FF_CALL send with no retry -- if it is
* lost (the worker briefly disconnects, or the controller it was talking
* to fails over to a new primary mid-flight), the job's true outcome is
* never recorded anywhere and its state (RUN) sticks forever. This test
* models that starting condition directly: a fresh persist state in which
* the job is durably recorded as RUN, assigned to a worker host, with NO
* finish fact of any kind ever appended for it -- exactly as if the real
* XCECLI_FF_CALL send had simply vanished before this feature ever
* existed. No second controller process or literal primary-to-primary
* handoff mechanism is simulated -- the OBSERVABLE starting condition (RUN
* job, no finish fact, worker has since actually finished it) is what
* this feature's own recovery mechanism must correct, regardless of how
* that condition came to exist; per Requirement 1.3, the controller's own
* trigger does not distinguish a first join, an ordinary reconnect, or a
* post-failover reconnect to a new primary -- the same rejoin path
* (`rejoin_worker`) is exercised below exactly as it would be for any of
* the three.
*
* Drives the SAME real rejoin -> query-send -> report-apply chain task
* 7.1's own round-trip test established (`rejoin_worker`, the REAL join
* entry point; `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) -- narrowed to ONE job, ONE outcome (finished-success),
* because this test's whole point is to isolate the bug-1741 scenario, not
* to duplicate the four-outcome coverage task 7.1 already built.
*
* finished-success stays on record_job_finish_fact_at's DEFERRED pattern
* (task 3.2/bug-1747) -- so, like test_full_rejoin_reconciliation_round_
* trip's own Phase 1/Phase 2 split, this test needs a real restart-replay
* (Phase 2, the SAME apply_committed_entry() seam a live controller's own
* commit-apply pump -- or another controller instance recovering the same
* durable log -- would invoke) to observe the job actually reach DONE;
* this bare-persist test fixture has no live commit-engine apply pump of
* its own, exactly the situation task 7.1's own Implementation Notes
* document. */
#define TWR_JOBID_FAILOVER_LOST 81401ULL
static void test_failover_during_job_recovers_lost_finish(void)
{
char tmp[256];
uint64_t worker_reported_finish_time_s;
strcpy(tmp, "/tmp/xce_twr72_failover_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
/* Deliberately far in the past -- the job genuinely finished well
* before this reconciliation round runs, so a reconciliation-
* processing-time bug (time(NULL)) could never accidentally produce a
* matching value. Same defensive idiom as this file's other finish-
* time tests. */
worker_reported_finish_time_s = (uint64_t)time(NULL) - 5400;
/* Phase 1: the (new) primary's own persist state -- job RUN, no
* finish fact anywhere in it (the lost-notify starting condition) --
* then the REAL rejoin/query/report chain. */
{
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* job;
hostattrs_t attrs, new_attrs;
xce_hostconfig_t hconfig;
endpoint_t stack_ep;
endpoint_t* ep;
xce_request_t decoded_query, decoded_report;
int rc;
uint64_t lsn_before, lsn_after;
int i;
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*)"twr72-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*)"twr72-controller";
c.myhost_info.key = (char*)"twr72-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 a durable WAL_JOB_SUBMIT fact + a matching in-memory job_t,
* RUN, assigned to the worker host -- and NOTHING ELSE. No
* WAL_JOB_START/DISPATCH/FINISH is ever appended for this job in
* this log: the controller's own durable state simply has no
* record the job ever finished, exactly bug-1741's own reported
* gap. */
{
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_FAILOVER_LOST;
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 = TWR_JOBID_FAILOVER_LOST;
jdef.state = JOB_STAT_RUN;
jdef.submit = &submit;
job = makeJob(c.jobmgr, &jdef);
TEST_ASSERT_NOT_NULL(job);
job->job.allocs = twr_make_test_alloc_for_host(
TWR_JOBID_FAILOVER_LOST, "resync-failover-host");
TEST_ASSERT_NOT_NULL(job->job.allocs);
}
/* ── Worker rejoin: the REAL join entry point. Requirement 1.3 --
* the controller does not need to know or care whether this is
* the host's first-ever join, an ordinary network reconnect, or
* (this scenario) a reconnect to a new primary after a controller
* failover; the same trigger fires because the controller still
* holds one non-terminal job for this host. */
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-failover-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 finding this host's one lost-finish job still "
"recorded as RUN must arm a reconciliation round (Req 1.1, "
"1.3)");
TEST_ASSERT_EQUAL_INT(1, w->resync->n_queried_jobids);
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(1,
decoded_query.body.jobstatequerymsg->n_total,
"the REAL wire query must ask about exactly the one job the "
"controller still believes is running (Req 2.1)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(TWR_JOBID_FAILOVER_LOST,
decoded_query.body.jobstatequerymsg->jobids[0],
"the query must identify the job unambiguously by its ID "
"(Req 2.2)");
xce_free_request(&decoded_query);
/* ── The worker's REAL answer: it actually finished this job
* successfully, long before this rejoin -- the controller simply
* never heard about it (the lost XCECLI_FF_CALL). */
{
xce_job_state_entry_t e;
xce_job_state_entry_t* entries[1];
xce_job_state_report_t report;
e = (xce_job_state_entry_t){
.jobid = TWR_JOBID_FAILOVER_LOST,
.status = JobStateStatus_finishedOk,
.exit_code = 0,
.finish_time_ms = worker_reported_finish_time_s * 1000ULL,
.queried = 1
};
entries[0] = &e;
report = (xce_job_state_report_t){
.n_total = 1, .pos = 0, .count = 1, .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);
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");
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: the round resolves and a durable fact
* capturing the worker's TRUE outcome is appended -- but the job
* is NOT yet observably DONE in this process's own memory. That
* is expected and correct: finished-success reconciles through
* record_job_finish_fact_at's DEFERRED pattern (task 3.2/bug-
* 1747), same as an ordinary finish report -- the state mutation
* happens later, when the fact is applied (Phase 2 below), never
* synchronously here. */
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, job->job.state,
"finished-success stays RUN synchronously -- the DEFERRED "
"record_job_finish_fact_at pattern applies the state change "
"later, via apply_committed_entry (Phase 2)");
TEST_ASSERT_NULL_MESSAGE(w->resync,
"the round's only queried job has a reconciled outcome -- it "
"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");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_twr_update_hook_fires,
"finished-success is DEFERRED -- no HOOK_DETAIL_JOB_UPDATE "
"fire is expected synchronously here; that fire is proven "
"below, in Phase 2, once the durable fact actually applies");
TEST_ASSERT_TRUE_MESSAGE(lsn_after >= lsn_before + 1,
"the worker's true finished-success outcome must be durably "
"appended (WAL_JOB_FINISH) as part of this report, even "
"though it has not applied to in-memory state yet");
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, persist_recover() -- the
* SAME apply_committed_entry() seam a live controller's own commit-
* apply pump (or another controller instance recovering this same
* durable log) would invoke. THIS is the test's core, observable
* completion criterion: the job the controller's own durable state
* showed stuck in RUN a moment ago, with its finish notify never
* having reached any controller instance, now reconstructs as
* genuinely finished -- DONE, with the worker's own true finish time,
* not reconciliation-processing wall-clock time -- because the
* worker rejoined and this feature's reconciliation round recovered
* it (Req 4.1, 4.2). */
{
persist_t* p2 = twr_make_fresh_persist(tmp);
cluster_t c2;
job_t* j2 = NULL;
int rc;
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_FAILOVER_LOST, (void**)&j2);
unlock(c2.jobmgr);
TEST_ASSERT_NOT_NULL_MESSAGE(j2,
"restart/rejoin-to-new-primary replay must reconstruct the "
"job from its WAL_JOB_SUBMIT fact");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j2->job.state,
"the feature's core completion criterion (bug-1741): a job "
"whose finish notify never reached any controller instance "
"ends up correctly finished, not stuck in RUN, once the "
"worker rejoins the new primary (Req 4.1)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(worker_reported_finish_time_s,
(uint64_t)j2->job.finish_time,
"the recovered finish_time must be the WORKER-reported "
"finish_time_ms from the reconciliation report, not "
"reconciliation-processing or replay wall-clock time "
"(Req 4.2, 3.5)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_twr_update_hook_fires,
"replaying the durable fact this reconciliation round "
"produced must reach the scheduler's/apiserver's notify path "
"(HOOK_DETAIL_JOB_UPDATE) exactly once for this one real "
"state change (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_failover_during_job_recovers_lost_finish: PASS\n");
}
/* ── Test group entry point ───────────────────────────────────────────── */
void test_worker_resync(void)
@ -3036,4 +3465,5 @@ void test_worker_resync(void)
RUN_TEST(test_handle_report_page_stale_round_discarded_when_superseded);
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);
}