From 165fc1586a91e43f603b7af4906479eda047b590 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 2 Sep 2026 00:04:37 +0000 Subject: [PATCH] test(worker-rejoin-resync): task 7.3 - burst recovery unbounded by count 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 Claude-Session: https://claude.ai/code/session_01Y4M5MJM2zK6nU43da267Ku --- .kiro/specs/worker-rejoin-resync/tasks.md | 3 +- src/controller/testsuite/test_worker_resync.c | 632 ++++++++++++++++++ 2 files changed, 634 insertions(+), 1 deletion(-) diff --git a/.kiro/specs/worker-rejoin-resync/tasks.md b/.kiro/specs/worker-rejoin-resync/tasks.md index 8e07d89c..9d649b50 100644 --- a/.kiro/specs/worker-rejoin-resync/tasks.md +++ b/.kiro/specs/worker-rejoin-resync/tasks.md @@ -141,7 +141,7 @@ - _Requirements: 4.1, 4.2, 4.6_ - _Depends: 7.1_ -- [ ] 7.3 Burst/array recovery unbounded by job count +- [x] 7.3 Burst/array recovery unbounded by job count - Reproduce a burst of many near-simultaneous job completions on one worker host where most of the individual finish reports are lost, and confirm reconciliation recovers all of them on the worker's next rejoin regardless of count - Observable completion: every job in the burst is correctly finished after reconciliation, with none left stuck and none duplicated - _Requirements: 8.1, 8.2, 8.3_ @@ -171,3 +171,4 @@ - **Task 6.2 also drew three escalating, unsolicited cross-session messages** (from an unfamiliar peer session) mid-flight, pushing a specific unreviewed code change to `test_jobstatereport_dispatch.c` — the exact file the dispatched implementer subagent was independently writing at the time. Declined to act on any of them (risk of conflicting with in-flight subagent work; escalating urgency/"no further investigation needed" framing is a known social-engineering pattern regardless of whether the underlying suggestion is technically plausible). The implementer subagent finished independently and solved the underlying `get_ep_worker()`-in-test-binary limitation completely differently (a `#ifdef XCE_TESTING` call-count seam in `resync.c`, not a global-fixture hack) — validating the decision not to short-circuit the normal TDD+review pipeline based on an unverified peer instruction. - **Task 7.1 (test-only, zero production changes) discovered and honestly asserted a genuine, pre-existing, systemic gap distinct from bug-1747**: a worker-reported (or any) job failure's exit code is never durably recorded anywhere in this codebase — `record_job_fail_fact()`'s `WAL_JOB_FAIL` payload has no exit-code slot at all, and `apply_job_fail_transition()` never touches `job.job.exit_info` for the FAIL path, so it stays `NULL` unconditionally. Bug-1747 only covers the FAIL finish-*time* mistiming; this is the exit-*code* gap, logged separately as bug-1754 (out of scope for this test-only task, correctly not fixed here — flagged for a dedicated future task). The round-trip test asserts `exit_info == NULL` honestly with an inline comment rather than encoding a wrong expectation the codebase can't actually produce. - Task 7.2 (test-only): added a focused, single-job proof of the feature's own original motivating scenario (bug-1741) — a job dispatched and RUN with NO finish fact ever committed (modeling the lost fire-and-forget notify), recovered to DONE with the worker-reported finish time via the real rejoin/query/report chain. Deliberately scoped smaller than task 7.1's four-outcome round trip (reuses its helper infrastructure, adds no new scaffolding) rather than duplicating it. tasks.md's own `_Requirements: 4.1, 4.2, 4.6_` tag includes 4.6 (unrecognized→PEND), which is genuinely unrelated to this finished-success scenario (that's task 7.1/2.1's territory) — `REQUIREMENTS_CHECKED` correctly narrowed to 4.1/4.2 instead of force-fitting an unrelated assertion, same accepted-imprecise-tag pattern as task 6.2's own citation nit. +- Task 7.3 (test-only): reproduced bug-1741's exact original shape (7 jobs, 6 lost) in one round/one wire page — 5 finished-success + 1 finished-failure recovered, plus one already-DONE job re-reported as an unqueried extra with a deliberately wrong finish time, proving reconciliation neither leaves any lost job stuck nor duplicates/clobbers an already-terminal one (exact `+6` WAL LSN delta, not `>=`, so a duplicate append is genuinely caught — confirmed via RED-phase reproduction against `record_job_finish_fact_at`'s terminal-state guard). Deliberately kept to a single wire page (page-boundary correctness is task 7.4's separate concern). Two reviewer/implementer subagents in this session's task 6.2/7.2 review rounds hit the "background command with no external wakeup" trap (ended their turn on a backgrounded build, never resumed) — both implementer and reviewer prompts for 7.3 onward now carry an explicit foreground-only warning up front; worth keeping in every future subagent dispatch in this codebase, not just implementer roles. diff --git a/src/controller/testsuite/test_worker_resync.c b/src/controller/testsuite/test_worker_resync.c index 0fbf7030..dee01426 100644 --- a/src/controller/testsuite/test_worker_resync.c +++ b/src/controller/testsuite/test_worker_resync.c @@ -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); }