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

test(worker-rejoin-resync): task 7.1 - full rejoin round trip

Browse source 
Add test_full_rejoin_reconciliation_round_trip proving the whole
chain composes end-to-end in one round, for a mix of still-running,
finished-success, finished-failure, and unrecognized jobs: real
rejoin_worker() -> worker_send_resync_query() (real wire query) ->
hand-built worker answer through the real report codec ->
resync_handle_report_page() -> resync_apply_report(), asserting
correct final state and scheduler notification for each. Test-only,
no production changes.

Logs bug-1754: a worker-reported job failure's exit code is never
durably recorded anywhere in this codebase (distinct from bug-1747's
finish-time gap) - discovered honestly asserted by the new test,
out of scope for this validation-only task.

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 21:47:22 +00:00
parent 9f655c61c7
commit 8650cfd9ba
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
3 changed files with 601 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -51,7 +51,16 @@
#include "hook.h"
#include "jobj.h"
#include "kvmgr.h"
#include "netio.h" /* worker-rejoin-resync task 7.1: endpoint_get/
* endpoint_deq/ITPTR_TO_PACKET, for real wire capture of
* the round-trip test's query send / report send */
#include "persist.h"
#include "request.h" /* worker-rejoin-resync task 7.1: xce_decode,
* xce_free_request, xce_job_state_report_req */
#include "resobj.h" /* worker-rejoin-resync task 7.1: makeResource/
* xce_resource_t, for the round-trip test's real
* resmgr (apply_join_state, reached via rejoin_worker,
* needs one) */
#include "resync.h" /* worker-rejoin-resync task 3.1 */
#include "wal_types.h"
#include "wobj.h" /* worker-rejoin-resync task 3.1: makeWorker, for a real
@ -62,6 +71,21 @@
#include "xcealloc.h"
#include "xstring.h"
/* worker-rejoin-resync task 7.1: rejoin_worker/worker_send_resync_query are
* non-static in worker.c when XCE_TESTING is defined -- forward-declared
* directly here rather than via a header, mirroring test_worker_join.c's
* own identical forward declarations exactly (these functions have no
* public header of their own; test_worker_join.c already establishes this
* as the file's own convention for driving the real join path from a
* test). collect_confvars (needed for worker.c to link, since on_worker_
* join -> worker_push_config -> collect_confvars) is already provided by
* test_worker_join.c elsewhere in this SAME controller_unit_test binary --
* not redefined here to avoid a duplicate-symbol link error; this test
* never calls worker_push_config anyway. */
int rejoin_worker(cluster_t* c, worker_t* w, endpoint_t* ep,
hostattrs_t* attrs, xce_hostconfig_t* hconfig);
int worker_send_resync_query(cluster_t* c, worker_t* w, endpoint_t* ep);
/* jobj.c's job type table and the process-global xstring table are shared
* across every test file in the controller_unit_test binary -- guard
* idempotently, mirroring every sibling fixture in this suite
@ -2435,6 +2459,554 @@ static void test_handle_report_page_malformed_bounds_rejected_and_recovers(
" PASS\n");
}
/* ══════════════════════════════════════════════════════════════════════
* Task 7.1 (Req 1.1-1.3, 2.1-2.2, 3.1-3.3, 3.6, 4.1-4.8, 7.1-7.3): full
* rejoin-reconciliation round trip against a live controller+worker pair.
*
* design.md's own Testing Strategy files this under "Integration Tests" as
* "against a live controller+worker pair, mirroring test_worker_join.c's
* fixture conventions (setup_test_cluster/teardown_test_cluster, direct
* calls, one test per behavior)" -- NOT separate OS processes or real TCP
* daemons. test_jobstatereport_dispatch.c's own header comment (task 6.2)
* additionally confirms a genuine wire dispatch through myreq_handler /
* handle_jobstatereport() cannot resolve a REAL worker_t in THIS binary at
* all: get_ep_worker() (misc.c) is not linked into controller_unit_test --
* a fixed test double defined in test_commit_barrier_reply.c stands in for
* it, always returning NULL/a fixture-controlled stub, for every test file
* in this binary, this one included. So this test drives the exact same
* REAL production entry points test_worker_join.c's own task 6.1 tests use
* directly -- rejoin_worker() (the real join path) to arm the round and
* worker_send_resync_query() to send the real wire query -- then feeds a
* REAL wire-encoded-then-decoded OPCode_jobStateReport (via task 1.4's own
* xce_job_state_report_req()/xce_decode() codec) into resync_handle_report_
* page() (task 6.2's own dispatch/reassembly entry point -- never resync_
* apply_report() directly), exactly mirroring how a real worker's answer
* would arrive and be reassembled/dispatched, without needing get_ep_
* worker()'s (unavailable) wire-to-worker_t resolution.
*
* Proves COMPOSITION, not per-outcome correctness -- each individual outcome
* (still-running no-op, finished-success, finished-failure, unrecognized)
* is already exhaustively unit-tested in isolation by tasks 2.1/2.2/3.2 (see
* test_apply_report_all_five_outcomes_updates_job_table above, and task
* 6.1's own test_worker_join.c tests for arm+query-send in isolation). This
* test's own value-add is proving all of join -> arm -> real wire query ->
* real wire report -> dispatch/reassembly -> apply -> scheduler-notify
* compose correctly TOGETHER, for a MIX of four different job outcomes, in
* ONE round, in ONE test run -- and that the scheduler notify path is
* reached for every actual state change produced, across BOTH of this
* feature's two distinct notify patterns: the EAGER path (unrecognized ->
* pending, task 2.1's pattern -- notified synchronously within the same
* call, asserted in Phase 1 below) and the DEFERRED path (finished-success/
* failure, task 3.2's pattern -- notified only once the durable fact is
* later applied). This bare-persist RETAIN-mode fixture has no live commit-
* engine apply pump to do that second half automatically, so this test
* proves it the same way test_apply_report_finished_success_uses_worker_
* reported_finish_time and test_reconcile_assign_survives_restart_replay
* already establish as this file's own precedent for that exact situation:
* a genuine two-phase restart-replay (Phase 1: real cluster + disk persist,
* drive the real chain, flush, close; Phase 2: fresh persist_t + fresh
* cluster_t, persist_recover() -- the SAME apply_committed_entry() seam a
* live controller's commit-apply pump would invoke). */
#define TWR_JOBID_RT_RUNNING 81301ULL
#define TWR_JOBID_RT_FINOK 81302ULL
#define TWR_JOBID_RT_FINFAIL 81303ULL
#define TWR_JOBID_RT_UNREC 81304ULL
#define TWR_RT_N_RES 9
/* Mirrors test_worker_join.c's own new_test_endpoint/pop_and_decode/assert_
* queue_empty helpers exactly -- real (heap) endpoint_t via endpoint_get()
* (bug-378: never a hand-rolled/zeroed stack stand-in for anything actually
* sent/received over), for genuine wire capture of both the query this test
* sends TO the worker and the report it simulates receiving FROM one. */
static endpoint_t* twr_rt_new_test_endpoint(void)
{
endpoint_t* ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
return ep;
}
static void twr_rt_pop_and_decode(endpoint_t* ep, xce_request_t* out)
{
data_item_t* it = endpoint_deq(ep);
TEST_ASSERT_NOT_NULL(it);
io_packet_t* pkt = ITPTR_TO_PACKET(it);
memset(out, 0, sizeof(*out));
TEST_ASSERT_EQUAL_INT(0, xce_decode(pkt, out)); /* consumes/frees pkt */
}
static void test_full_rejoin_reconciliation_round_trip(void)
{
char tmp[256];
uint64_t worker_reported_finish_time_s;
int32_t worker_reported_exit_code = 42;
strcpy(tmp, "/tmp/xce_twr71_roundtrip_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
/* Deliberately far from "now" so a reconciliation-processing-time bug
* (time(NULL)) could never accidentally produce a matching value --
* same defensive idiom test_apply_report_finished_success_uses_worker_
* reported_finish_time already establishes. */
worker_reported_finish_time_s = (uint64_t)time(NULL) - 3600;
/* Phase 1: real cluster (workermgr + resmgr + jobmgr + hooksys) + real
* disk persist -- drives the REAL join/query-send/report-apply chain.
* Own local cluster_t (not the shared g_twr fixture), mirroring every
* other restart-replay test in this file, since Phase 1's disk state
* must survive past Phase 1's own teardown into Phase 2 below. */
{
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* jobs[4];
uint64_t jobids[4] = { TWR_JOBID_RT_RUNNING,
TWR_JOBID_RT_FINOK,
TWR_JOBID_RT_FINFAIL,
TWR_JOBID_RT_UNREC };
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[4] = { 0, 0, 0, 0 };
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*)"twr71-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*)"twr71-controller";
c.myhost_info.key = (char*)"twr71-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
* for all 4 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). All 4 are non-terminal, so all 4 are
* expected in the round's query set (Req 2.1) -- "unrecognized" is
* not a DIFFERENT starting state from the controller's own point of
* view, only a different worker ANSWER. */
for (i = 0; i < 4; 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 = jobids[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 = jobids[i];
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(
jobids[i], "resync-roundtrip-host");
TEST_ASSERT_NOT_NULL(jobs[i]->job.allocs);
}
/* ── Join: rejoin_worker (the REAL join entry point) arms the round,
* worker_send_resync_query sends the REAL wire query (Req 1.1-1.3,
* 2.1, 2.2). */
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-roundtrip-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 4 known non-terminal jobs on this host must arm "
"a reconciliation round (Req 1.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(4, w->resync->n_queried_jobids,
"the round's query set must match exactly the host's 4 "
"non-terminal jobs (Req 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(4,
decoded_query.body.jobstatequerymsg->n_total,
"the REAL wire query (Req 2.1, 2.2) must list exactly the 4 "
"armed jobids");
for (i = 0; i < decoded_query.body.jobstatequerymsg->count; i++) {
uint64_t id = decoded_query.body.jobstatequerymsg->jobids[i];
if (id == TWR_JOBID_RT_RUNNING) saw[0] = 1;
if (id == TWR_JOBID_RT_FINOK) saw[1] = 1;
if (id == TWR_JOBID_RT_FINFAIL) saw[2] = 1;
if (id == TWR_JOBID_RT_UNREC) saw[3] = 1;
}
TEST_ASSERT_TRUE_MESSAGE(saw[0] && saw[1] && saw[2] && saw[3],
"the wire query must cover all 4 jobs by unique ID (Req 2.2)");
xce_free_request(&decoded_query);
/* ── Worker's answer: a REAL wire-encoded-then-decoded OPCode_
* jobStateReport (task 1.4's own xce_job_state_report_req()/xce_
* decode() codec), covering all 4 outcomes in one page, fed into
* resync_handle_report_page() -- task 6.2's own dispatch/
* reassembly entry point, never resync_apply_report() directly
* (Req 3.1-3.3, 3.6, 8.1-8.3). */
{
xce_job_state_entry_t e_running, e_finok, e_finfail, e_unrec;
xce_job_state_entry_t* entries[4];
xce_job_state_report_t report;
e_running = (xce_job_state_entry_t){
.jobid = TWR_JOBID_RT_RUNNING,
.status = JobStateStatus_running,
.exit_code = 0, .finish_time_ms = 0, .queried = 1
};
e_finok = (xce_job_state_entry_t){
.jobid = TWR_JOBID_RT_FINOK,
.status = JobStateStatus_finishedOk,
.exit_code = 0,
.finish_time_ms = worker_reported_finish_time_s * 1000ULL,
.queried = 1
};
e_finfail = (xce_job_state_entry_t){
.jobid = TWR_JOBID_RT_FINFAIL,
.status = JobStateStatus_finishedFail,
.exit_code = worker_reported_exit_code,
.finish_time_ms = 0, .queried = 1
};
e_unrec = (xce_job_state_entry_t){
.jobid = TWR_JOBID_RT_UNREC,
.status = JobStateStatus_unrecognized,
.exit_code = 0, .finish_time_ms = 0, .queried = 1
};
entries[0] = &e_running;
entries[1] = &e_finok;
entries[2] = &e_finfail;
entries[3] = &e_unrec;
report = (xce_job_state_report_t){
.n_total = 4, .pos = 0, .count = 4, .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);
/* ── End-to-end assertions (Phase 1 -- everything observable before
* any commit round trip, mirroring record_job_reconcile_pending_
* fact's own eager-mutation precedent). */
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, jobs[0]->job.state,
"still-running (queried) must be left exactly as-is -- no fact "
"recorded (Req 4.1)");
TEST_ASSERT_NOT_NULL(jobs[0]->job.allocs);
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, jobs[1]->job.state,
"finished-success stays in the DEFERRED record_job_finish_"
"fact_at pattern -- state mutates later via apply_committed_"
"entry (proven below, Phase 2), not synchronously here, exactly "
"as test_apply_report_all_five_outcomes_updates_job_table's own "
"established precedent documents");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, jobs[2]->job.state,
"finished-failure stays in the DEFERRED on_job_fail/record_job_"
"fail_fact pattern -- same deferred precedent as finished-"
"success");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, jobs[3]->job.state,
"unrecognized reconciles EAGERLY through record_job_reconcile_"
"pending_fact (Req 4.5, 4.6) -- PEND synchronously, before any "
"commit round trip");
TEST_ASSERT_NULL_MESSAGE(jobs[3]->job.allocs,
"bug-1631: reconciled-to-pending must release the allocation "
"synchronously");
TEST_ASSERT_NULL_MESSAGE(w->resync,
"once every one of the 4 queried jobs 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");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_twr_update_hook_fires,
"Req 7.1, 7.2, 7.3: the EAGER unrecognized-to-pending "
"transition must reach the scheduler's/apiserver's notify path "
"SYNCHRONOUSLY, within this same call -- the two DEFERRED "
"outcomes (finished-success/failure) have not applied yet at "
"this point, so they must NOT have notified yet either "
"(confirmed to notify later, in Phase 2 below)");
TEST_ASSERT_TRUE_MESSAGE(lsn_after >= lsn_before + 3,
"at least 3 durable facts must be appended for this report: "
"finished-success (WAL_JOB_FINISH), finished-failure (WAL_JOB_"
"FAIL), and unrecognized (WAL_JOB_UPDATE) -- the still-running "
"entry records none");
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 two DEFERRED outcomes above. Mirrors
* test_apply_report_finished_success_uses_worker_reported_finish_
* time's own established two-phase idiom exactly. */
{
persist_t* p2 = twr_make_fresh_persist(tmp);
cluster_t c2;
job_t* j2_running = NULL;
job_t* j2_finok = NULL;
job_t* j2_finfail = NULL;
job_t* j2_unrec = 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_RT_RUNNING, (void**)&j2_running);
key_find(c2.jobmgr, (void*)TWR_JOBID_RT_FINOK, (void**)&j2_finok);
key_find(c2.jobmgr, (void*)TWR_JOBID_RT_FINFAIL, (void**)&j2_finfail);
key_find(c2.jobmgr, (void*)TWR_JOBID_RT_UNREC, (void**)&j2_unrec);
unlock(c2.jobmgr);
TEST_ASSERT_NOT_NULL(j2_running);
TEST_ASSERT_NOT_NULL(j2_finok);
TEST_ASSERT_NOT_NULL(j2_finfail);
TEST_ASSERT_NOT_NULL(j2_unrec);
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j2_running->job.state,
"still-running: no fact was ever recorded for it, so replay "
"leaves it exactly as WAL_JOB_SUBMIT alone reconstructs it "
"(Req 4.1)");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j2_finok->job.state,
"finished-success: the DEFERRED record_job_finish_fact_at fact "
"must apply to DONE on replay (Req 4.1)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(worker_reported_finish_time_s,
(uint64_t)j2_finok->job.finish_time,
"Req 4.2, 3.5: the recovered finish_time must be the WORKER-"
"reported finish_time_ms, not reconciliation-processing "
"wall-clock time");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_EXIT, j2_finfail->job.state,
"finished-failure: the DEFERRED on_job_fail/record_job_fail_"
"fact fact must apply to EXIT on replay (Req 4.1)");
/* Req 4.2 is DOCUMENTED-unsatisfiable for the finished-FAILURE
* finish_time specifically (bug-1747, pre-existing, systemic,
* explicitly deferred, unrelated to this task's own boundary) --
* NOT re-asserted here as if it were correct. This round-trip test
* additionally confirms, and CONCERNS reports, a related, more
* complete gap bug-1747's own text does not mention: the worker-
* reported EXIT CODE (entry->exit_code, plumbed all the way into
* job_failinfo_t.code by resync_apply_finished_entry, resync.c) is
* not merely mistimed but is NEVER durably recorded or applied to
* job.job.exit_info AT ALL for a FAIL outcome, by any existing path
* in this codebase -- record_job_fail_fact's own WAL_JOB_FAIL
* payload (persist_submit_job_fact_ex(..., WAL_JOB_FAIL, jobid, 0,
* WAL_JOB_FAIL_STAGE_FLAG|stage), job.c) carries no code slot at
* all, and apply_job_fail_transition() (job.c) never touches job.
* job.exit_info. This is asserted honestly below as NULL -- the
* TRUE current behavior -- rather than asserting a "correct" value
* this codebase cannot actually produce (this file's own task 3.2
* Implementation Notes: "a green suite is not proof of correctness
* when the test itself encodes the wrong expectation"). */
TEST_ASSERT_NULL_MESSAGE(j2_finfail->job.exit_info,
"pre-existing, systemic gap (see comment above, reported in "
"this task's own CONCERNS, not fixed here): a worker-reported "
"exit code for a FAIL outcome is never durably recorded by any "
"existing path in this codebase -- exit_info stays NULL, not "
"the worker-reported code");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j2_unrec->job.state,
"unrecognized: the EAGER record_job_reconcile_pending_fact's "
"durable WAL_JOB_UPDATE fact must ALSO reconstruct correctly on "
"replay -- convergence via either the live eager path (Phase 1 "
"above) or a restart, not just the live path (Req 4.5, 4.6)");
TEST_ASSERT_NULL(j2_unrec->job.allocs);
TEST_ASSERT_EQUAL_INT_MESSAGE(3, 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 (HOOK_DETAIL_JOB_UPDATE) for EVERY actual state change -- "
"finished-success, finished-failure, and unrecognized-to-"
"pending -- exactly 3 times (still-running recorded no fact, "
"and WAL_JOB_SUBMIT fires HOOK_DETAIL_JOB_NEW, not JOB_UPDATE, "
"so it does not add to this count). Together with Phase 1's own "
"single synchronous fire, this confirms EVERY actual state "
"change this round produced -- both the eager and the deferred "
"ones -- reaches the same notify path, not just some of them");
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_full_rejoin_reconciliation_round_trip: PASS\n");
}
/* ── Test group entry point ───────────────────────────────────────────── */
void test_worker_resync(void)
@ -2463,4 +3035,5 @@ void test_worker_resync(void)
RUN_TEST(test_handle_report_page_multi_page_reassembles_and_applies_once);
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);
}