/* test_wal_job_preason_fact.c — bug-1692 regression suite * * Proves WAL_JOB_PREASON's payload -- previously bare wal_jobstate_t{jobid, * state}, exactly the same silent-field-drop bug class as bug-1678 * (WAL_JOB_DISPATCH/job.allocs), bug-1687 (WAL_JOB_START/job.pidinfo+ * start_time), bug-1689 (WAL_JOB_FINISH/job.exit_info+finish_time), bug-1690 * (WAL_JOB_RUSAGE/job.run_usage), and bug-1691 (WAL_JOB_CKPT/job.ckptinfo) -- * now carries the scheduler-reported host/resource-based pending reasons * (job.pend_reasons, xce_job_preason_t*) through BOTH of persist.c's apply * seams: * * - apply_committed_entry() (live-commit apply / boot replay dispatch) * - apply_wal_entry() (standby catch-up / replicated-apply dispatch) * * The JobPendReason/HostBasedReason/ResourceBasedReason Cap'n Proto structs * ALREADY existed in controller.capnp (used elsewhere for the full-Job * snapshot/checkpoint path) and matched xce_job_preason_t/xce_host_reason_t/ * xce_res_reason_t field-for-field, so NO schema change was needed -- this * fix reuses the already-generated encode_JobPendReason/decode_JobPendReason/ * free_JobPendReason_ptr functions directly, mirroring bug-1678's * WAL_JOB_DISPATCH/JobAlloc reuse pattern. * * IMPORTANT DIVERGENCE FROM WAL_JOB_START/FINISH/DISPATCH's regression * suites, SAME AS bug-1690/1691: set_preason()/do_clr_preason() (job.c) * never change job.state -- persist_job_hook_fun's HOOK_DETAIL_SETPREASON/ * HOOK_DETAIL_CLRPREASON arm captures js.state verbatim from the job's * CURRENT (unchanged) state, so apply_job_update_transition() reports * JOB_TRANSITION_NOOP_DUPLICATE on every single real PREASON replay. Gating * pend_reasons adoption on result==JOB_TRANSITION_APPLIED would therefore * make it dead code for the realistic, only-ever-occurring case. * test_wjp_apply_committed_entry_sets_preason_no_state_change below is this * suite's most important test. * * IMPORTANT DIVERGENCE FROM bug-1690/1691's OWN precedent: unlike * ckptinfo/run_usage, job.pend_reasons has a genuine CLEAR operation * (do_clr_preason, HOOK_DETAIL_CLRPREASON) that legitimately resets it back * to NULL and must survive replay just as durably as the SET side. CKPT's/ * RUSAGE's convention ("payload of exactly sizeof(wal_jobstate_t) == no * trailing bytes == leave field untouched") cannot distinguish that * legitimate NULL from an old-format payload, so this fix adds an explicit * presence byte after the wal_jobstate_t header (0 = cleared, adopt NULL; 1 * = capnp JobPendReason bytes follow). test_wjp_apply_committed_entry_ * clears_preason / test_wjp_apply_wal_entry_clears_preason / * test_wjp_real_clear_survives_restart below are the tests that would have * silently failed under a naive copy of the CKPT/RUSAGE wire shape. * * ALSO DIVERGES FROM bug-1690's WAL_JOB_RUSAGE fix (but matches bug-1691's * WAL_JOB_CKPT): adoption is a plain free-and-replace (via the capnp- * generated free_JobPendReason_ptr()), NOT a merge. job.pend_reasons and its * nested host/resource reason arrays are plain calloc()/strdup()-backed * throughout -- verified via erase_job_preason()'s FREEUP-only release * (src/lib/preason.c) and every add_*_preason() constructor (strdup()/ * calloc()), plus the capnp-generated decode_HostBasedReason/ * decode_ResourceBasedReason using STRING_DUP+calloc (src/lib/ * controller.capnp.c) -- NOT xstring/RTTI-backed like ckptinfo/rusage's * host_name, so no merge-not-replace hazard exists here. * * Also proves backward compatibility: a WAL_JOB_PREASON entry with the OLD * payload shape (exactly sizeof(wal_jobstate_t), no presence byte -- * simulating an entry written before this fix) still decodes cleanly, with * pend_reasons left untouched exactly as before this fix -- no crash, no * misread. * * Fixtures mirror test_wal_job_ckpt_fact.c's own conventions exactly: * - wjp_setup/wjp_teardown: bare cluster_t (jobmgr + hooksys) + in-memory * persist_t. * - wjp_make_fresh_persist/wjp_teardown_persist: real disk-backed * persist_t + persist_recover(), for restart-replay tests. * * test_wjp_real_do_set_preason_survives_restart / test_wjp_real_clear_ * survives_restart drive the REAL, unmodified do_set_preason()/ * do_clr_preason() (the actual scheduler-reported pend-reason receipt * handlers) through a real WAL-write + persist_recover() restart cycle -- * unlike on_job_ckpt()/on_job_finish(), these need a real (if minimal, * zeroed, no-iomgr) endpoint_t because they call xce_reply() before * processing, mirroring test_commit_barrier_reply.c's own established * "zeroed static endpoint_t, fd=-1, packets land on ep->outq" pattern * (drained here via endpoint_deq()/free_packet() to avoid a reachable-leak * false positive under the suite's LeakSanitizer pass). set_preason() itself * is file-static to job.c, so do_set_preason() -- the real production * entrypoint (job.c's only exported wrapper) -- is what these tests drive, * exactly as the scheduler's actual OPCode_setpreason/OPCode_clrpreason * wire path does. */ #include #include #include #include #include #include #include #include #include #include #include "cluster.h" #include "controller.capnp.h" #include "hook.h" #include "jobj.h" #include "persist.h" #include "preason.h" #include "typemap.h" #include "wal_types.h" #include "xce/cluster_def.h" #include "xce/types.h" #include "xcealloc.h" #include "xstring.h" /* jobj.c's jobsub 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. */ static int g_wjp_jobobj_ready; static cluster_t g_wjp_c; static persist_t g_wjp_p; /* ── Spy hook: counts AFTER_EVENT firings, ignores all payload args ─────── */ static int g_wjp_hook_fires; static int wjp_spy_hook(int type, int situation, int detail, void* data, va_list ap) { (void)type; (void)detail; (void)data; (void)ap; if (situation == HOOK_AFTER_EVENT) g_wjp_hook_fires++; return 0; } static void wjp_setup(void) { kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; if (makeString((char*)"wjp-probe") == NULL) xstring_table_init(64); if (!g_wjp_jobobj_ready) { TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); g_wjp_jobobj_ready = 1; } memset(&g_wjp_c, 0, sizeof(g_wjp_c)); TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&g_wjp_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); g_wjp_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); TEST_ASSERT_NOT_NULL(g_wjp_c.jobmgr); memset(&g_wjp_p, 0, sizeof(g_wjp_p)); g_wjp_p.cluster = &g_wjp_c; g_wjp_hook_fires = 0; } static void wjp_teardown(void) { xce_hook_fini(&g_wjp_c.hooksys); if (g_wjp_c.jobmgr) dereference(g_wjp_c.jobmgr); } /* Look up a job_t wrapper by jobid (NULL if absent). */ static job_t* wjp_find_job(uint64_t jobid) { job_t* j = NULL; lock(g_wjp_c.jobmgr, RTTI_LOCK_READ); int found = key_find(g_wjp_c.jobmgr, (void*)jobid, (void**)&j); unlock(g_wjp_c.jobmgr); return (found > 0) ? j : NULL; } /* Creates a job directly in the fixture's jobmgr, bypassing dispatch/replay * entirely — for seeding fact-transition tests, mirroring * test_wal_job_ckpt_fact.c's wjc_seed_job. */ static job_submit_data_t g_wjp_seed_submit; static job_t* wjp_seed_job(uint64_t jobid, uint32_t state) { xce_job_t jdef; memset(&g_wjp_seed_submit, 0, sizeof(g_wjp_seed_submit)); g_wjp_seed_submit.queue = (char*)"q"; g_wjp_seed_submit.user = (char*)"tester"; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = jobid; jdef.state = state; jdef.submit = &g_wjp_seed_submit; return makeJob(g_wjp_c.jobmgr, &jdef); } /* Build an xce_job_preason_t with one host-based AND one resource-based * reason (exercises both nested list types, per this bug's audit * requirement) — caller owns the result (release via * free_JobPendReason_ptr, mirroring persist.c's own adoption verb, or * erase_job_preason+FREEUP for a manually-built one). */ static xce_job_preason_t* wjp_build_preason(int main_reason, const char* host_name, int host_reason, const char* res_name, const char* res_host, int res_reason) { xce_job_preason_t* pr = calloc(1, sizeof(*pr)); TEST_ASSERT_NOT_NULL(pr); pr->main_reason = main_reason; TEST_ASSERT_TRUE( add_job_preason(pr, host_reason, (char*)host_name, NULL) > 0); TEST_ASSERT_TRUE( add_job_preason(pr, res_reason, (char*)res_host, (char*)res_name) > 0); return pr; } static void wjp_assert_preason_matches(job_t* j, int main_reason, const char* host_name, int host_reason, const char* res_name, const char* res_host, int res_reason) { xce_job_preason_t* pr = j->job.pend_reasons; TEST_ASSERT_NOT_NULL_MESSAGE(pr, "a WAL_JOB_PREASON payload with a trailing capnp blob must set " "job.pend_reasons (bug-1692), not just state"); TEST_ASSERT_EQUAL_INT(main_reason, pr->main_reason); TEST_ASSERT_EQUAL_INT_MESSAGE(1, pr->nhost_reasons, "host-based reason list must survive replay"); TEST_ASSERT_NOT_NULL(pr->host_based_reasons); TEST_ASSERT_NOT_NULL(pr->host_based_reasons[0]); TEST_ASSERT_EQUAL_STRING(host_name, pr->host_based_reasons[0]->host); TEST_ASSERT_EQUAL_INT(host_reason, pr->host_based_reasons[0]->reason); TEST_ASSERT_EQUAL_INT_MESSAGE(1, pr->nres_reasons, "resource-based reason list must survive replay"); TEST_ASSERT_NOT_NULL(pr->res_based_reasons); TEST_ASSERT_NOT_NULL(pr->res_based_reasons[0]); TEST_ASSERT_EQUAL_STRING(res_name, pr->res_based_reasons[0]->resource); TEST_ASSERT_EQUAL_STRING(res_host, pr->res_based_reasons[0]->host); TEST_ASSERT_EQUAL_INT(res_reason, pr->res_based_reasons[0]->reason); } /* Local mirror of persist.c's static capnp_encode_jobpreason() -- this * translation unit cannot reach that file-static function, exactly like * test_wal_job_ckpt_fact.c's wjc_encode_job mirrors capnp_encode_job. */ static int wjp_encode_jobpreason(const xce_job_preason_t* pr, uint8_t** out, size_t* out_len) { struct capn c; struct capn_segment* cs; struct JobPendReason cjp; JobPendReason_ptr jpp; capn_init_malloc(&c); cs = capn_root(&c).seg; encode_JobPendReason(cs, &cjp, (xce_job_preason_t*)pr); jpp = new_JobPendReason(cs); write_JobPendReason(&cjp, jpp); capn_setp(capn_root(&c), 0, jpp.p); *out_len = capn_size(&c); *out = malloc(*out_len); if (!*out) { capn_free(&c); return -ENOMEM; } capn_write_mem(&c, *out, *out_len, 0); capn_free(&c); return 0; } /* Build a wal_jobstate_t header + presence byte (1) + trailing capnp * JobPendReason bytes — persist_job_hook_fun's WAL_JOB_PREASON "set" payload * shape (bug-1692). Caller owns *out. */ static void wjp_build_set_payload(uint64_t jobid, uint32_t state, const xce_job_preason_t* pr, uint8_t** out, size_t* out_len) { wal_jobstate_t js = { .jobid = jobid, .state = state, ._pad = 0 }; uint8_t* capnp_buf = NULL; size_t capnp_len = 0; TEST_ASSERT_EQUAL_INT(0, wjp_encode_jobpreason(pr, &capnp_buf, &capnp_len)); *out_len = sizeof(js) + 1 + capnp_len; *out = malloc(*out_len); TEST_ASSERT_NOT_NULL(*out); memcpy(*out, &js, sizeof(js)); (*out)[sizeof(js)] = 1; if (capnp_len > 0) memcpy(*out + sizeof(js) + 1, capnp_buf, capnp_len); free(capnp_buf); } /* Build a wal_jobstate_t header + presence byte (0), no trailing bytes — * persist_job_hook_fun's WAL_JOB_PREASON "clear" payload shape (bug-1692, * do_clr_preason's own fact). Caller owns *out. */ static void wjp_build_clear_payload(uint64_t jobid, uint32_t state, uint8_t** out, size_t* out_len) { wal_jobstate_t js = { .jobid = jobid, .state = state, ._pad = 0 }; *out_len = sizeof(js) + 1; *out = malloc(*out_len); TEST_ASSERT_NOT_NULL(*out); memcpy(*out, &js, sizeof(js)); (*out)[sizeof(js)] = 0; } /* ── apply_committed_entry(): live-commit apply / boot replay dispatch ──── */ static void test_wjp_apply_committed_entry_sets_preason_no_state_change(void) { wjp_setup(); job_t* j = wjp_seed_job(75001, JOB_STAT_PEND); TEST_ASSERT_NOT_NULL(j); xce_job_preason_t* pr = wjp_build_preason(7, "worker-3", 11, "gpu", "worker-4", 12); uint8_t* payload = NULL; size_t plen = 0; wjp_build_set_payload(75001, JOB_STAT_PEND, pr, &payload, &plen); erase_job_preason(pr); FREEUP(pr); xce_add_hook(&g_wjp_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjp_spy_hook, NULL); int rc = apply_committed_entry(&g_wjp_p, WAL_JOB_PREASON, 400, payload, (uint32_t)plen); free(payload); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); wjp_assert_preason_matches(j, 7, "worker-3", 11, "gpu", "worker-4", 12); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wjp_hook_fires, "a same-state (duplicate) PREASON replay -- the realistic, " "only-ever-occurring case, since set_preason() never changes " "job.state -- must NOT fire a notification hook, unchanged from " "before this fix, even though pend_reasons WAS adopted"); wjp_teardown(); printf( "test_wjp_apply_committed_entry_sets_preason_no_state_change: " "PASS\n"); } /* Secondary scenario: a WAL_JOB_PREASON fact that DOES carry a forward state * transition (mechanism-level test, mirroring bug-1690/1691's own coverage * of this same divergence). pend_reasons must be adopted here too, and the * existing forward-transition hook-fire behavior must be unchanged. */ static void test_wjp_apply_committed_entry_sets_preason_forward_transition(void) { wjp_setup(); job_t* j = wjp_seed_job(75002, JOB_STAT_PEND); TEST_ASSERT_NOT_NULL(j); xce_job_preason_t* pr = wjp_build_preason(8, "worker-5", 21, "mem", "worker-6", 22); uint8_t* payload = NULL; size_t plen = 0; wjp_build_set_payload(75002, JOB_STAT_RUN, pr, &payload, &plen); erase_job_preason(pr); FREEUP(pr); xce_add_hook(&g_wjp_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjp_spy_hook, NULL); int rc = apply_committed_entry(&g_wjp_p, WAL_JOB_PREASON, 401, payload, (uint32_t)plen); free(payload); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); wjp_assert_preason_matches(j, 8, "worker-5", 21, "mem", "worker-6", 22); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_wjp_hook_fires, "a forward PREASON-carried transition must still fire exactly one " "notification, unchanged from before this fix"); wjp_teardown(); printf( "test_wjp_apply_committed_entry_sets_preason_forward_transition: " "PASS\n"); } static void test_wjp_apply_committed_entry_backward_compat_old_payload(void) { wjp_setup(); job_t* j = wjp_seed_job(75003, JOB_STAT_PEND); TEST_ASSERT_NOT_NULL(j); TEST_ASSERT_NULL_MESSAGE(j->job.pend_reasons, "fixture must seed a job with pend_reasons still unset"); wal_jobstate_t js = { .jobid = 75003, .state = JOB_STAT_PEND, ._pad = 0 }; int rc = apply_committed_entry(&g_wjp_p, WAL_JOB_PREASON, 402, &js, sizeof(js)); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); TEST_ASSERT_NULL_MESSAGE(j->job.pend_reasons, "an old-format (exactly sizeof(wal_jobstate_t)) WAL_JOB_PREASON " "payload must leave pend_reasons NULL -- no crash, no misread of " "absent trailing bytes"); wjp_teardown(); printf("test_wjp_apply_committed_entry_backward_compat_old_payload: " "PASS\n"); } /* PREASON-specific (no CKPT/RUSAGE analog): a presence byte of 0 must * genuinely clear an existing pend_reasons on replay -- proving the * presence-byte design actually closes the gap a naive copy of bug-1690/ * 1691's "trailing bytes present or absent" convention would have missed. */ static void test_wjp_apply_committed_entry_clears_preason(void) { wjp_setup(); job_t* j = wjp_seed_job(75004, JOB_STAT_PEND); TEST_ASSERT_NOT_NULL(j); j->job.pend_reasons = wjp_build_preason(9, "stale-host", 1, "stale-res", "stale-host2", 2); uint8_t* payload = NULL; size_t plen = 0; wjp_build_clear_payload(75004, JOB_STAT_PEND, &payload, &plen); int rc = apply_committed_entry(&g_wjp_p, WAL_JOB_PREASON, 403, payload, (uint32_t)plen); free(payload); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_NULL_MESSAGE(j->job.pend_reasons, "a presence-byte=0 (explicit clear) WAL_JOB_PREASON payload must " "reset pend_reasons to NULL on replay, not merely leave it " "untouched (bug-1692's key divergence from bug-1690/1691's " "precedent)"); wjp_teardown(); printf("test_wjp_apply_committed_entry_clears_preason: PASS\n"); } /* present=1 with ZERO trailing bytes (a malformed length no real encoder * ever produces -- capnp_encode_jobpreason() always emits at least a * minimal root-pointer+struct-header blob) must NOT be treated as a clear * (present != 0) and must NOT crash -- decode_jobpreason() is never even * called (guarded by the length check), so `decoded` stays NULL and the * prior value must survive untouched, matching every other decode helper's * failure fallback in apply_committed_entry. (NOTE: unlike a length check, * capnp-c's read path does not itself validate content -- genuinely * malformed-but-nonempty bytes decode into a zero-valued struct rather than * failing, matching decode_joballoc/decode_jobrunusage's identical * leniency, so that scenario is not distinguishable from a real empty * JobPendReason at this layer; WAL entry CRC framing is what protects * against on-disk corruption, not this logical decode step.) */ static void test_wjp_apply_committed_entry_decode_failure_leaves_untouched(void) { wjp_setup(); job_t* j = wjp_seed_job(75005, JOB_STAT_PEND); TEST_ASSERT_NOT_NULL(j); j->job.pend_reasons = wjp_build_preason(4, "keep-host", 5, "keep-res", "keep-host2", 6); wal_jobstate_t js = { .jobid = 75005, .state = JOB_STAT_PEND, ._pad = 0 }; uint8_t payload[sizeof(js) + 1]; memcpy(payload, &js, sizeof(js)); payload[sizeof(js)] = 1; /* present, but no capnp bytes follow at all */ int rc = apply_committed_entry(&g_wjp_p, WAL_JOB_PREASON, 404, payload, (uint32_t)sizeof(payload)); TEST_ASSERT_EQUAL_INT(0, rc); wjp_assert_preason_matches(j, 4, "keep-host", 5, "keep-res", "keep-host2", 6); wjp_teardown(); printf( "test_wjp_apply_committed_entry_decode_failure_leaves_untouched: " "PASS\n"); } /* ── apply_wal_entry(): standby catch-up / replicated-apply dispatch ────── */ static void test_wjp_apply_wal_entry_sets_preason_no_hook(void) { wjp_setup(); job_t* j = wjp_seed_job(75006, JOB_STAT_PEND); TEST_ASSERT_NOT_NULL(j); xce_job_preason_t* pr = wjp_build_preason(3, "worker-7", 31, "disk", "worker-8", 32); uint8_t* payload = NULL; size_t plen = 0; wjp_build_set_payload(75006, JOB_STAT_PEND, pr, &payload, &plen); erase_job_preason(pr); FREEUP(pr); xce_add_hook(&g_wjp_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjp_spy_hook, NULL); int rc = apply_wal_entry(&g_wjp_p, WAL_JOB_PREASON, 405, payload, (uint32_t)plen); free(payload); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); wjp_assert_preason_matches(j, 3, "worker-7", 31, "disk", "worker-8", 32); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wjp_hook_fires, "apply_wal_entry's catch-up dispatch must never fire a " "notification hook"); wjp_teardown(); printf("test_wjp_apply_wal_entry_sets_preason_no_hook: PASS\n"); } static void test_wjp_apply_wal_entry_backward_compat_old_payload(void) { wjp_setup(); job_t* j = wjp_seed_job(75007, JOB_STAT_PEND); TEST_ASSERT_NOT_NULL(j); TEST_ASSERT_NULL_MESSAGE(j->job.pend_reasons, "fixture must seed a job with pend_reasons still unset"); wal_jobstate_t js = { .jobid = 75007, .state = JOB_STAT_PEND, ._pad = 0 }; int rc = apply_wal_entry(&g_wjp_p, WAL_JOB_PREASON, 406, &js, sizeof(js)); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); TEST_ASSERT_NULL_MESSAGE(j->job.pend_reasons, "an old-format WAL_JOB_PREASON payload must leave pend_reasons " "NULL on this seam too"); wjp_teardown(); printf("test_wjp_apply_wal_entry_backward_compat_old_payload: PASS\n"); } static void test_wjp_apply_wal_entry_clears_preason(void) { wjp_setup(); job_t* j = wjp_seed_job(75008, JOB_STAT_PEND); TEST_ASSERT_NOT_NULL(j); j->job.pend_reasons = wjp_build_preason(2, "stale-a", 1, "stale-b", "stale-c", 2); uint8_t* payload = NULL; size_t plen = 0; wjp_build_clear_payload(75008, JOB_STAT_PEND, &payload, &plen); int rc = apply_wal_entry(&g_wjp_p, WAL_JOB_PREASON, 407, payload, (uint32_t)plen); free(payload); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_NULL_MESSAGE(j->job.pend_reasons, "a presence-byte=0 WAL_JOB_PREASON payload must clear " "pend_reasons on the standby catch-up seam too"); wjp_teardown(); printf("test_wjp_apply_wal_entry_clears_preason: PASS\n"); } /* ── Restart-replay observables ──────────────────────────────────────── */ static persist_t* wjp_make_fresh_persist(const char* tmpdir) { persist_t* p = calloc(1, sizeof(persist_t)); if (!p) return NULL; persist_config_t cfg = persist_config_defaults(); cfg.wal_sync_ms = 5000; /* slow fsync — no races in tests */ persist_set_workdir(p, tmpdir, "ctrl-test"); if (persist_init(p, &cfg, NULL) != 0) { free(p); return NULL; } if (persist_start(p) != 0) { persist_fini(p); free(p); return NULL; } return p; } static void wjp_teardown_persist(persist_t* p) { persist_stop(p); persist_fini(p); free(p); } /* Encode helper mirroring test_wal_job_ckpt_fact.c's wjc_encode_job * (capnp_encode_job is file-static in persist.c). */ static int wjp_encode_job(const xce_job_t* j, uint8_t** out, size_t* out_len) { struct capn c; struct capn_segment* cs; struct Job cj; Job_ptr jp; capn_init_malloc(&c); cs = capn_root(&c).seg; encode_Job(cs, &cj, (xce_job_t*)j); jp = new_Job(cs); write_Job(&cj, jp); capn_setp(capn_root(&c), 0, jp.p); *out_len = capn_size(&c); *out = malloc(*out_len); if (!*out) { capn_free(&c); return -ENOMEM; } capn_write_mem(&c, *out, *out_len, 0); capn_free(&c); return 0; } /* Direct-payload restart replay: appends WAL_JOB_SUBMIT + a new-format * WAL_JOB_PREASON (real host+resource reasons) straight onto disk via * persist_wal_async, then replays via a fresh persist_t's persist_recover() * (a real restart, exercising apply_committed_entry's boot-replay dispatch). * Isolates the WAL fact shape from the live caller, mirroring * test_wal_job_ckpt_fact.c's own restart-replay test. */ static void test_wjp_restart_replay_recovers_preason(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_wjp_replay_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); /* Phase 1: write the log to real disk. */ { persist_t* p = wjp_make_fresh_persist(tmp); TEST_ASSERT_NOT_NULL(p); job_submit_data_t submit; memset(&submit, 0, sizeof(submit)); submit.queue = (char*)"q"; submit.user = (char*)"tester"; xce_job_t jdef; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = 75009; jdef.state = JOB_STAT_PEND; jdef.submit = &submit; uint8_t* job_payload = NULL; size_t job_plen = 0; TEST_ASSERT_EQUAL_INT(0, wjp_encode_job(&jdef, &job_payload, &job_plen)); TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_SUBMIT, job_payload, (uint32_t)job_plen)); free(job_payload); xce_job_preason_t* pr = wjp_build_preason(6, "replay-host", 41, "replay-res", "replay-host2", 42); uint8_t* pr_payload = NULL; size_t pr_plen = 0; wjp_build_set_payload(75009, JOB_STAT_PEND, pr, &pr_payload, &pr_plen); erase_job_preason(pr); FREEUP(pr); TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_PREASON, pr_payload, (uint32_t)pr_plen)); free(pr_payload); TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); wjp_teardown_persist(p); } /* Phase 2: fresh persist_t + fresh cluster_t, recover from disk — * simulates a controller restart. */ { persist_t* p2 = wjp_make_fresh_persist(tmp); TEST_ASSERT_NOT_NULL(p2); wjp_setup(); p2->cluster = &g_wjp_c; int rc = persist_recover(p2); TEST_ASSERT_EQUAL_INT(0, rc); job_t* j = wjp_find_job(75009); TEST_ASSERT_NOT_NULL_MESSAGE(j, "restart replay must construct the job via WAL_JOB_SUBMIT"); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); wjp_assert_preason_matches(j, 6, "replay-host", 41, "replay-res", "replay-host2", 42); wjp_teardown(); wjp_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_wjp_restart_replay_recovers_preason: PASS\n"); } /* Drain an endpoint's reply queue (avoids a reachable-leak false positive * under this suite's LeakSanitizer pass — mirrors test_commit_barrier_ * reply.c's cbrp_scan_replies, minus the reply-content decode this suite * doesn't need). */ static void wjp_drain_ep(endpoint_t* ep) { data_item_t* it; while ((it = endpoint_deq(ep)) != NULL) { io_packet_t* pkt = ITPTR_TO_PACKET(it); free_packet(pkt); free(pkt); } } /* bug-1692's primary regression test: drives the REAL production * do_set_preason() path (the actual scheduler-reported pend-reason receipt * handler — job.c's exported entrypoint for OPCode_setpreason; set_preason() * itself is file-static) with a real xce_job_preason_t, through a real * WAL-write + persist_recover() restart cycle — the audit's exact confirmed * root cause. do_set_preason() calls xce_reply() before processing, so a * minimal zeroed (fd=-1, no iomgr) endpoint_t is needed, mirroring * test_commit_barrier_reply.c's established fixture pattern. */ static void test_wjp_real_do_set_preason_survives_restart(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_wjp_onset_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); const uint64_t jobid = 75010; /* Phase 1: real cluster + real disk persist. Seed WAL_JOB_SUBMIT * directly (so replay can reconstruct the job) AND a live in-memory job * object (so the real do_set_preason() call site has something to * find/lock) -- mirrors test_wjc_real_on_job_ckpt_survives_restart's * dual-seed pattern exactly. Then drive an actual do_set_preason() call * and let IT append WAL_JOB_PREASON via persist_job_hook_fun -- this is * the production code path under test, not a hand-built WAL entry. */ { persist_config_t cfg; if (makeString((char*)"wjp-onset-probe") == NULL) xstring_table_init(64); if (!g_wjp_jobobj_ready) { TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); g_wjp_jobobj_ready = 1; } cluster_t c; kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; memset(&c, 0, sizeof(c)); TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&c.hooksys, XCE_CONTROLLER_HOOK_MAX)); c.jobmgr = makeInstance(&xkvmgr_type, &int_params); TEST_ASSERT_NOT_NULL(c.jobmgr); c.myhost = &c.myhost_info; c.myhost_info.name = (char*)"ctrl-test"; c.myhost_info.key = (char*)""; cfg = persist_config_defaults(); cfg.wal_sync_ms = 5000; persist_set_workdir(&c.persist, tmp, "ctrl-test"); /* Same reasoning as test_wjc_real_on_job_ckpt_survives_restart: * do_set_preason()/set_preason() (job.c) are hook-driven and rely * ENTIRELY on persist_job_hook_fun being registered as a c.hooksys * subscriber to reach the WAL at all -- persist_start() only * registers that hook `if (p->cluster != NULL)`. */ TEST_ASSERT_EQUAL_INT(0, persist_init(&c.persist, &cfg, &c)); TEST_ASSERT_EQUAL_INT(0, persist_start(&c.persist)); /* Durable WAL_JOB_SUBMIT seed. */ { job_submit_data_t wal_submit; xce_job_t wal_jdef; uint8_t* buf = NULL; size_t sz = 0; memset(&wal_submit, 0, sizeof(wal_submit)); wal_submit.queue = (char*)"oq"; wal_submit.user = (char*)"tester"; memset(&wal_jdef, 0, sizeof(wal_jdef)); wal_jdef.jobid = jobid; wal_jdef.state = JOB_STAT_PEND; wal_jdef.submit = &wal_submit; TEST_ASSERT_EQUAL_INT(0, wjp_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); } /* Live in-memory job object for do_set_preason()/set_preason() to * find/lock. */ { job_submit_data_t submit; xce_job_t jdef; memset(&submit, 0, sizeof(submit)); submit.queue = (char*)"oq"; submit.user = (char*)"tester"; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = jobid; jdef.state = JOB_STAT_PEND; jdef.submit = &submit; TEST_ASSERT_NOT_NULL(makeJob(c.jobmgr, &jdef)); } /* The real scheduler-reported pend-reason set. jpr is heap- * allocated and MUST be, since set_preason() MOVE_PTRs it straight * into the job (job.c) -- mirrors test_commit_barrier_reply.c's * cbrp_arrange_set_preason's own doc comment on this exact point. */ xce_job_preason_t* jpr = wjp_build_preason(5, "real-host", 51, "real-res", "real-host2", 52); xce_pendreason_t pend = { .jobid = jobid, .reasons = jpr }; xce_pendreason_t* prs[1] = { &pend }; xce_preason_set_t set; memset(&set, 0, sizeof(set)); set.nreasons = 1; set.reasons = prs; xce_peerinfo_t from = { .name = (char*)"scheduler", .key = (char*)"" }; xce_request_t req; memset(&req, 0, sizeof(req)); req.opcode = OPCode_setpreason; req.seq = 1; req.from = &from; req.body_which = Request_body_preasonmsg; req.body.preasonmsg = &set; endpoint_t ep; memset(&ep, 0, sizeof(ep)); ep.fd = -1; int rc = do_set_preason(&c, &ep, &req); TEST_ASSERT_EQUAL_INT(0, rc); wjp_drain_ep(&ep); TEST_ASSERT_EQUAL_INT(0, persist_flush(&c.persist)); persist_stop(&c.persist); persist_fini(&c.persist); dereference(c.jobmgr); xce_hook_fini(&c.hooksys); } /* Phase 2: fresh persist_t + fresh cluster_t (jobmgr/hooksys only), * recover from disk — simulates a controller restart. */ { persist_t* p2 = wjp_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); p2->cluster = &c2; rc = persist_recover(p2); TEST_ASSERT_EQUAL_INT(0, rc); lock(c2.jobmgr, RTTI_LOCK_READ); key_find(c2.jobmgr, (void*)jobid, (void**)&j2); unlock(c2.jobmgr); TEST_ASSERT_NOT_NULL_MESSAGE(j2, "restart replay must reconstruct the job via WAL_JOB_SUBMIT"); TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j2->job.state, "restart replay must recover the job's state, unchanged by the " "real do_set_preason() call"); /* bug-1692: restart replay must recover the scheduler-reported * pend_reasons -- the audit's exact confirmed root cause (mirrors * bug-1678/1687/1689/1690/1691's own acceptance criterion). */ wjp_assert_preason_matches(j2, 5, "real-host", 51, "real-res", "real-host2", 52); xce_hook_fini(&c2.hooksys); if (c2.jobmgr) dereference(c2.jobmgr); wjp_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_wjp_real_do_set_preason_survives_restart: PASS\n"); } /* PREASON-specific (no CKPT/RUSAGE analog): drives the REAL do_set_preason() * THEN the REAL do_clr_preason() before the restart, proving the clear * genuinely survives replay too -- not just the set. This is the scenario a * naive copy of bug-1690/1691's "trailing bytes present or absent" wire * convention would have silently gotten wrong (a stale, pre-clear * pend_reasons would resurrect on restart). */ static void test_wjp_real_clear_survives_restart(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_wjp_onclr_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); const uint64_t jobid = 75011; { persist_config_t cfg; if (makeString((char*)"wjp-onclr-probe") == NULL) xstring_table_init(64); if (!g_wjp_jobobj_ready) { TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); g_wjp_jobobj_ready = 1; } cluster_t c; kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; memset(&c, 0, sizeof(c)); TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&c.hooksys, XCE_CONTROLLER_HOOK_MAX)); c.jobmgr = makeInstance(&xkvmgr_type, &int_params); TEST_ASSERT_NOT_NULL(c.jobmgr); c.myhost = &c.myhost_info; c.myhost_info.name = (char*)"ctrl-test"; c.myhost_info.key = (char*)""; 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, &c)); TEST_ASSERT_EQUAL_INT(0, persist_start(&c.persist)); { job_submit_data_t wal_submit; xce_job_t wal_jdef; uint8_t* buf = NULL; size_t sz = 0; memset(&wal_submit, 0, sizeof(wal_submit)); wal_submit.queue = (char*)"oq"; wal_submit.user = (char*)"tester"; memset(&wal_jdef, 0, sizeof(wal_jdef)); wal_jdef.jobid = jobid; wal_jdef.state = JOB_STAT_PEND; wal_jdef.submit = &wal_submit; TEST_ASSERT_EQUAL_INT(0, wjp_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); } { job_submit_data_t submit; xce_job_t jdef; memset(&submit, 0, sizeof(submit)); submit.queue = (char*)"oq"; submit.user = (char*)"tester"; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = jobid; jdef.state = JOB_STAT_PEND; jdef.submit = &submit; TEST_ASSERT_NOT_NULL(makeJob(c.jobmgr, &jdef)); } endpoint_t ep; memset(&ep, 0, sizeof(ep)); ep.fd = -1; /* SET first (real path), then CLEAR (real path) -- both must be * durably reflected by the time we restart below. */ { xce_job_preason_t* jpr = wjp_build_preason(1, "temp-host", 61, "temp-res", "temp-host2", 62); xce_pendreason_t pend = { .jobid = jobid, .reasons = jpr }; xce_pendreason_t* prs[1] = { &pend }; xce_preason_set_t set; memset(&set, 0, sizeof(set)); set.nreasons = 1; set.reasons = prs; xce_peerinfo_t from = { .name = (char*)"scheduler", .key = (char*)"" }; xce_request_t req; memset(&req, 0, sizeof(req)); req.opcode = OPCode_setpreason; req.seq = 1; req.from = &from; req.body_which = Request_body_preasonmsg; req.body.preasonmsg = &set; TEST_ASSERT_EQUAL_INT(0, do_set_preason(&c, &ep, &req)); wjp_drain_ep(&ep); } { uint64_t jobs[1] = { jobid }; xce_clear_reason_t clr = { .njobs = 1, .jobs = jobs }; xce_peerinfo_t from = { .name = (char*)"scheduler", .key = (char*)"" }; xce_request_t req; memset(&req, 0, sizeof(req)); req.opcode = OPCode_clrpreason; req.seq = 2; req.from = &from; req.body_which = Request_body_clrpreasonmsg; req.body.clrpreasonmsg = &clr; TEST_ASSERT_EQUAL_INT(0, do_clr_preason(&c, &ep, &req)); wjp_drain_ep(&ep); } TEST_ASSERT_EQUAL_INT(0, persist_flush(&c.persist)); persist_stop(&c.persist); persist_fini(&c.persist); dereference(c.jobmgr); xce_hook_fini(&c.hooksys); } { persist_t* p2 = wjp_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); p2->cluster = &c2; rc = persist_recover(p2); TEST_ASSERT_EQUAL_INT(0, rc); lock(c2.jobmgr, RTTI_LOCK_READ); key_find(c2.jobmgr, (void*)jobid, (void**)&j2); unlock(c2.jobmgr); TEST_ASSERT_NOT_NULL_MESSAGE(j2, "restart replay must reconstruct the job via WAL_JOB_SUBMIT"); /* bug-1692's key correctness fix: the CLEAR (do_clr_preason) must * durably survive replay, not just the earlier SET -- a stale * pend_reasons resurrecting here would be exactly the bug a naive * copy of bug-1690/1691's wire convention would have produced. */ TEST_ASSERT_NULL_MESSAGE(j2->job.pend_reasons, "restart replay must recover the CLEARED pend_reasons state " "(NULL), not resurrect the earlier SET's stale value"); xce_hook_fini(&c2.hooksys); if (c2.jobmgr) dereference(c2.jobmgr); wjp_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_wjp_real_clear_survives_restart: PASS\n"); } /* ── Test group entry point ───────────────────────────────────────────── */ void test_wal_job_preason_fact(void) { UnitySetTestFile(__FILE__); RUN_TEST(test_wjp_apply_committed_entry_sets_preason_no_state_change); RUN_TEST(test_wjp_apply_committed_entry_sets_preason_forward_transition); RUN_TEST(test_wjp_apply_committed_entry_backward_compat_old_payload); RUN_TEST(test_wjp_apply_committed_entry_clears_preason); RUN_TEST(test_wjp_apply_committed_entry_decode_failure_leaves_untouched); RUN_TEST(test_wjp_apply_wal_entry_sets_preason_no_hook); RUN_TEST(test_wjp_apply_wal_entry_backward_compat_old_payload); RUN_TEST(test_wjp_apply_wal_entry_clears_preason); RUN_TEST(test_wjp_restart_replay_recovers_preason); RUN_TEST(test_wjp_real_do_set_preason_survives_restart); RUN_TEST(test_wjp_real_clear_survives_restart); }