/* test_wal_job_rusage_fact.c — bug-1690 regression suite * * Proves WAL_JOB_RUSAGE'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) and bug-1689 (WAL_JOB_FINISH/job.exit_info+finish_time) -- * now carries the client-reported, accumulated per-host CPU/mem usage * (job.run_usage, an xce_job_runusage_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) * * SECOND, EQUALLY IMPORTANT DIVERGENCE: adoption is a MERGE * (merge_job_runusage(), job.c), never a free-and-replace. job.run_usage * can be the job's LIVE structure -- update_job_rusages()/update_host_ * runusage() (job.c/misc.c) mutate it synchronously and in place, with a * host_name sourced from makeString() (an RTTI-managed, header-before-data * pointer, NOT a plain malloc()) -- and the primary's own live apply pump * (persist_apply_pump_run()) re-applies that SAME fact, on that SAME live * job, moments after the live mutation already happened. Freeing it via * the capnp-generated free_JobRunusage_ptr() (written for plain malloc'd, * capnp-decoded Text fields) would free a non-malloc'd pointer -- heap * corruption under ASAN. test_wru_apply_committed_entry_merges_without_ * freeing_live_host below proves this directly with a real makeString()'d * host_name. * * IMPORTANT DIVERGENCE FROM WAL_JOB_START/FINISH/DISPATCH's regression * suites: a rusage report almost never changes job.state (the job stays RUN * across every repeat report) -- unlike a START/FINISH/DISPATCH, which are * genuine one-time PEND->RUN / RUN->terminal transitions. apply_committed_ * entry()'s shared job-fact group only stamps type-specific fields when * apply_job_update_transition() reports JOB_TRANSITION_APPLIED (a forward * state change) -- gating run_usage adoption on that same condition would * make it dead code for the realistic, common case (every report after the * first hits JOB_TRANSITION_NOOP_DUPLICATE). The fix decodes/adopts * run_usage UNCONDITIONALLY, before that gate. test_wru_apply_committed_ * entry_sets_runusage_no_state_change below is this suite's most important * test: it proves run_usage still lands even when state does NOT change -- * the scenario a naive copy of the START/FINISH/DISPATCH pattern would have * silently failed to cover. * * xce_job_runusage_t (xce/types.h) had NO existing Cap'n Proto mapping at * all -- unlike JobAlloc/JobPIDInfo, which bug-1678/1687 could reuse * standalone from the existing Job struct's own sub-fields. This fix adds a * genuine new schema pair, HostRunusage + JobRunusage (controller.capnp), * distinct from the similar-looking but unrelated xce_job_rusages_t/ * JobRusages/SchedJobRusages (the client's WIRE UPDATE REQUEST shape). * * Also proves backward compatibility: a WAL_JOB_RUSAGE entry with the OLD * payload shape (exactly sizeof(wal_jobstate_t), no trailing bytes -- * simulating an entry written before this fix, or one fired while * job.run_usage was still NULL at write time) still decodes cleanly, with * run_usage left untouched exactly as before this fix -- no crash, no * misread. * * Fixtures mirror test_wal_job_start_fact.c's own conventions exactly: * - wru_setup/wru_teardown: bare cluster_t (jobmgr + hooksys) + in-memory * persist_t. * - wru_make_fresh_persist/wru_teardown_persist: real disk-backed * persist_t + persist_recover(), for restart-replay tests. * * test_wru_real_hook_appends_and_survives_restart drives the REAL, * unmodified persist_job_hook_fun() (the actual production hook subscriber * this fix changed) via the "variadic trampoline" pattern established by * test_commit_barrier_rollout.c/test_persist_hooks.c, rather than going * through update_job_rusages() (job.c) itself: this same controller_unit_ * test binary already links test_commit_barrier_reply.c, which defines its * OWN non-static stub of update_host_runusage() (a no-op, since misc.c is * not part of controller_unit_test_src) -- calling update_job_rusages() * from this file would silently hit that no-op stub and never populate * job.run_usage at all, testing nothing new. Driving persist_job_hook_fun() * directly with a job_t whose run_usage is set on the fixture (bypassing * update_host_runusage entirely, which this fix does not touch) exercises * the actual changed function on the actual production code path a real * update_job_rusages() call reaches it from. */ #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 "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_wru_jobobj_ready; static cluster_t g_wru_c; static persist_t g_wru_p; /* ── Spy hook: counts AFTER_EVENT firings, ignores all payload args ─────── */ static int g_wru_hook_fires; static int wru_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_wru_hook_fires++; return 0; } static void wru_setup(void) { kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; if (makeString((char*)"wru-probe") == NULL) xstring_table_init(64); if (!g_wru_jobobj_ready) { TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); g_wru_jobobj_ready = 1; } memset(&g_wru_c, 0, sizeof(g_wru_c)); TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&g_wru_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); g_wru_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); TEST_ASSERT_NOT_NULL(g_wru_c.jobmgr); memset(&g_wru_p, 0, sizeof(g_wru_p)); g_wru_p.cluster = &g_wru_c; g_wru_hook_fires = 0; } static void wru_teardown(void) { xce_hook_fini(&g_wru_c.hooksys); if (g_wru_c.jobmgr) dereference(g_wru_c.jobmgr); } /* Look up a job_t wrapper by jobid (NULL if absent). */ static job_t* wru_find_job(uint64_t jobid) { job_t* j = NULL; lock(g_wru_c.jobmgr, RTTI_LOCK_READ); int found = key_find(g_wru_c.jobmgr, (void*)jobid, (void**)&j); unlock(g_wru_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_dispatch_fact.c's wjd_seed_job / test_wal_job_start_fact.c's * wjs_seed_job. */ static job_submit_data_t g_wru_seed_submit; static job_t* wru_seed_job(uint64_t jobid, uint32_t state) { xce_job_t jdef; memset(&g_wru_seed_submit, 0, sizeof(g_wru_seed_submit)); g_wru_seed_submit.queue = (char*)"q"; g_wru_seed_submit.user = (char*)"tester"; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = jobid; jdef.state = state; jdef.submit = &g_wru_seed_submit; return makeJob(g_wru_c.jobmgr, &jdef); } /* ── Real fixture data: 2 hosts' worth of accumulated CPU/mem usage ──────── */ static xce_host_runusage_t g_wru_host_a = { .host_name = (char*)"wru-host-a", .cpu_time = 1200, .mem_usage = 512.5 }; static xce_host_runusage_t g_wru_host_b = { .host_name = (char*)"wru-host-b", .cpu_time = 900, .mem_usage = 256.25 }; static xce_host_runusage_t* g_wru_host_ptrs[2]; static xce_job_runusage_t wru_make_test_runusage(void) { xce_job_runusage_t r; memset(&r, 0, sizeof(r)); r.max_mem = 1024.5; r.avg_mem = 768.25; r.used_mem = 900.0; r.cpu_time = 2100; r.peak_cpu_time = 1500; r.peak_cpu_duration = 30; r.mem_efficiency = 0.75f; r.avg_cpu_efficiency = 0.6f; r.peak_cpu_efficiency = 0.9f; r.num_hosts = 2; g_wru_host_ptrs[0] = &g_wru_host_a; g_wru_host_ptrs[1] = &g_wru_host_b; r.host_runusage = g_wru_host_ptrs; return r; } static void wru_assert_runusage_matches(xce_job_runusage_t* got) { TEST_ASSERT_NOT_NULL_MESSAGE(got, "a WAL_JOB_RUSAGE payload with trailing JobRunusage bytes must set " "job.run_usage (bug-1690), not just state"); TEST_ASSERT_EQUAL_FLOAT(1024.5, got->max_mem); TEST_ASSERT_EQUAL_FLOAT(768.25, got->avg_mem); TEST_ASSERT_EQUAL_FLOAT(900.0, got->used_mem); TEST_ASSERT_EQUAL_INT(2100, got->cpu_time); TEST_ASSERT_EQUAL_INT(1500, got->peak_cpu_time); TEST_ASSERT_EQUAL_INT(30, got->peak_cpu_duration); TEST_ASSERT_EQUAL_FLOAT(0.75f, got->mem_efficiency); TEST_ASSERT_EQUAL_FLOAT(0.6f, got->avg_cpu_efficiency); TEST_ASSERT_EQUAL_FLOAT(0.9f, got->peak_cpu_efficiency); TEST_ASSERT_EQUAL_INT(2, got->num_hosts); TEST_ASSERT_NOT_NULL(got->host_runusage); TEST_ASSERT_EQUAL_STRING("wru-host-a", got->host_runusage[0]->host_name); TEST_ASSERT_EQUAL_INT(1200, got->host_runusage[0]->cpu_time); TEST_ASSERT_EQUAL_FLOAT(512.5, got->host_runusage[0]->mem_usage); TEST_ASSERT_EQUAL_STRING("wru-host-b", got->host_runusage[1]->host_name); TEST_ASSERT_EQUAL_INT(900, got->host_runusage[1]->cpu_time); TEST_ASSERT_EQUAL_FLOAT(256.25, got->host_runusage[1]->mem_usage); } /* Build a wal_jobstate_t header + trailing Cap'n Proto JobRunusage bytes -- * persist_job_hook_fun's WAL_JOB_RUSAGE payload shape -- mirroring * wjs_build_start_payload() (test_wal_job_start_fact.c) since persist.c's * own static capnp_encode_jobrunusage() isn't linkable from a test TU. * Caller owns *out. */ static int wru_build_rusage_payload(uint64_t jobid, uint32_t state, xce_job_runusage_t* r, uint8_t** out, size_t* out_len) { struct capn c; struct capn_segment* cs; struct JobRunusage cjr; JobRunusage_ptr jrp; uint8_t* rbuf; size_t rlen; wal_jobstate_t js = { .jobid = jobid, .state = state, ._pad = 0 }; capn_init_malloc(&c); cs = capn_root(&c).seg; encode_JobRunusage(cs, &cjr, r); jrp = new_JobRunusage(cs); write_JobRunusage(&cjr, jrp); capn_setp(capn_root(&c), 0, jrp.p); rlen = capn_size(&c); rbuf = malloc(rlen); if (!rbuf) { capn_free(&c); return -ENOMEM; } capn_write_mem(&c, rbuf, rlen, 0); capn_free(&c); *out_len = sizeof(js) + rlen; *out = malloc(*out_len); if (!*out) { free(rbuf); return -ENOMEM; } memcpy(*out, &js, sizeof(js)); memcpy(*out + sizeof(js), rbuf, rlen); free(rbuf); return 0; } /* ── apply_committed_entry(): live-commit apply / boot replay dispatch ──── */ /* THE key regression test: seeds the job already RUNNING and replays a * WAL_JOB_RUSAGE fact whose own state is ALSO RUN -- the realistic, * overwhelmingly common shape of a real rusage report (job.state never * actually changes across repeat reports). apply_job_update_transition() * reports JOB_TRANSITION_NOOP_DUPLICATE for this (state == cur), so a fix * that (incorrectly) gated run_usage adoption on JOB_TRANSITION_APPLIED -- * mirroring WAL_JOB_START/FINISH/DISPATCH's own gated fields -- would leave * job.run_usage NULL here. This test fails against that naive shape and * passes against the actual fix (unconditional decode/adopt). */ static void test_wru_apply_committed_entry_sets_runusage_no_state_change(void) { wru_setup(); job_t* j = wru_seed_job(73001, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j); xce_job_runusage_t r = wru_make_test_runusage(); uint8_t* payload = NULL; size_t plen = 0; TEST_ASSERT_EQUAL_INT(0, wru_build_rusage_payload(73001, JOB_STAT_RUN, &r, &payload, &plen)); xce_add_hook(&g_wru_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wru_spy_hook, NULL); int rc = apply_committed_entry(&g_wru_p, WAL_JOB_RUSAGE, 300, payload, (uint32_t)plen); free(payload); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); wru_assert_runusage_matches(j->job.run_usage); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wru_hook_fires, "a same-state (duplicate) RUSAGE replay must NOT fire a " "notification hook -- unchanged from before this fix -- even though " "run_usage WAS adopted"); wru_teardown(); printf("test_wru_apply_committed_entry_sets_runusage_no_state_change: " "PASS\n"); } /* Secondary scenario: a WAL_JOB_RUSAGE fact that DOES carry a forward state * transition (mechanism-level test, mirroring how the START/DISPATCH suites * use PEND->RUN generically). run_usage must be adopted here too, and the * existing forward-transition hook-fire behavior must be unchanged. */ static void test_wru_apply_committed_entry_sets_runusage_forward_transition(void) { wru_setup(); job_t* j = wru_seed_job(73002, JOB_STAT_PEND); TEST_ASSERT_NOT_NULL(j); xce_job_runusage_t r = wru_make_test_runusage(); uint8_t* payload = NULL; size_t plen = 0; TEST_ASSERT_EQUAL_INT(0, wru_build_rusage_payload(73002, JOB_STAT_RUN, &r, &payload, &plen)); xce_add_hook(&g_wru_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wru_spy_hook, NULL); int rc = apply_committed_entry(&g_wru_p, WAL_JOB_RUSAGE, 301, payload, (uint32_t)plen); free(payload); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); wru_assert_runusage_matches(j->job.run_usage); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_wru_hook_fires, "a forward RUSAGE-carried transition must still fire exactly one " "notification, unchanged from before this fix"); wru_teardown(); printf( "test_wru_apply_committed_entry_sets_runusage_forward_transition: " "PASS\n"); } static void test_wru_apply_committed_entry_backward_compat_old_payload(void) { wru_setup(); job_t* j = wru_seed_job(73003, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j); TEST_ASSERT_NULL_MESSAGE(j->job.run_usage, "fixture must seed a job with run_usage still unset"); wal_jobstate_t js = { .jobid = 73003, .state = JOB_STAT_RUN, ._pad = 0 }; int rc = apply_committed_entry(&g_wru_p, WAL_JOB_RUSAGE, 302, &js, sizeof(js)); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); TEST_ASSERT_NULL_MESSAGE(j->job.run_usage, "an old-format (exactly sizeof(wal_jobstate_t)) WAL_JOB_RUSAGE " "payload must leave run_usage NULL -- no crash, no misread of " "absent trailing bytes"); wru_teardown(); printf("test_wru_apply_committed_entry_backward_compat_old_payload: " "PASS\n"); } /* ── Merge safety: never free a LIVE (possibly xstring-backed) host_name ── * * bug-1690's most important safety finding: update_job_rusages()/ * update_host_runusage() (job.c/misc.c) build the LIVE job.run_usage with * host_name sourced from makeString() (src/lib/xstring.c) -- an * RTTI-managed, header-before-data pointer (RTTI_DATA_TO_OBJ, librtti), NOT * a plain malloc() allocation. apply_committed_entry()'s WAL_JOB_RUSAGE * branch can run against this EXACT live structure -- the primary's own * live apply pump (persist_apply_pump_run()) re-applies a fact moments * after the fact's own live mutation already happened. A naive * free-and-replace (the capnp-generated free_JobRunusage_ptr(), written for * plain malloc'd Text fields) would free() a non-malloc'd pointer there: * guaranteed heap corruption under ASAN. This test builds a run_usage with * a REAL makeString()'d host_name (mirroring the live-mutation flavor * exactly), then applies a WAL_JOB_RUSAGE fact reporting an update for that * SAME host plus a brand-new second host, and asserts: (a) no crash * (implicit -- ASAN aborts mid-test on a corrupting free), (b) the existing * host's host_name POINTER IDENTITY survives (proving it was merged in * place, never freed+reallocated), (c) its cpu_time/mem_usage reflect the * new report, (d) the new host is correctly appended. */ static void test_wru_apply_committed_entry_merges_without_freeing_live_host(void) { wru_setup(); job_t* j = wru_seed_job(73008, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j); /* Build a LIVE-flavor run_usage by hand, mirroring update_host_ * runusage()'s own construction exactly: calloc'd struct/array/entries, * but a REAL makeString()'d host_name -- the exact allocator mismatch * this test exists to catch. */ char* live_host_name = makeString((char*)"wru-live-host"); TEST_ASSERT_NOT_NULL(live_host_name); xce_job_runusage_t* live = (xce_job_runusage_t*)calloc(1, sizeof(*live)); TEST_ASSERT_NOT_NULL(live); live->max_mem = 100.0; live->cpu_time = 50; live->num_hosts = 1; live->host_runusage = (xce_host_runusage_t**)calloc(1, sizeof(xce_host_runusage_t*)); TEST_ASSERT_NOT_NULL(live->host_runusage); live->host_runusage[0] = (xce_host_runusage_t*)calloc(1, sizeof(xce_host_runusage_t)); TEST_ASSERT_NOT_NULL(live->host_runusage[0]); live->host_runusage[0]->host_name = live_host_name; live->host_runusage[0]->cpu_time = 111; live->host_runusage[0]->mem_usage = 222.0; j->job.run_usage = live; /* Fact payload: an update for "wru-live-host" plus a brand-new * "wru-new-host" -- exactly what a real subsequent rusage report for * this job would carry. */ xce_host_runusage_t update_host = { .host_name = (char*)"wru-live-host", .cpu_time = 999, .mem_usage = 888.0 }; xce_host_runusage_t new_host = { .host_name = (char*)"wru-new-host", .cpu_time = 42, .mem_usage = 84.0 }; xce_host_runusage_t* host_ptrs[2] = { &update_host, &new_host }; xce_job_runusage_t r; memset(&r, 0, sizeof(r)); r.max_mem = 500.0; r.cpu_time = 250; r.num_hosts = 2; r.host_runusage = host_ptrs; uint8_t* payload = NULL; size_t plen = 0; TEST_ASSERT_EQUAL_INT(0, wru_build_rusage_payload(73008, JOB_STAT_RUN, &r, &payload, &plen)); int rc = apply_committed_entry(&g_wru_p, WAL_JOB_RUSAGE, 305, payload, (uint32_t)plen); free(payload); TEST_ASSERT_EQUAL_INT(0, rc); xce_job_runusage_t* got = j->job.run_usage; TEST_ASSERT_NOT_NULL(got); TEST_ASSERT_EQUAL_FLOAT(500.0, got->max_mem); TEST_ASSERT_EQUAL_INT(250, got->cpu_time); TEST_ASSERT_EQUAL_INT(2, got->num_hosts); int live_idx = -1, found_new = 0; for (int i = 0; i < got->num_hosts; i++) { xce_host_runusage_t* h = got->host_runusage[i]; TEST_ASSERT_NOT_NULL(h); if (h->host_name != NULL && strcmp(h->host_name, "wru-live-host") == 0) { live_idx = i; TEST_ASSERT_TRUE_MESSAGE(h->host_name == live_host_name, "an existing host's host_name pointer must be preserved by " "identity -- merge must never free+replace it (bug-1690 " "heap-corruption hazard: it may be an xstring, not a " "malloc'd pointer)"); TEST_ASSERT_EQUAL_INT(999, h->cpu_time); TEST_ASSERT_EQUAL_FLOAT(888.0, h->mem_usage); } else if (h->host_name != NULL && strcmp(h->host_name, "wru-new-host") == 0) { found_new = 1; TEST_ASSERT_EQUAL_INT(42, h->cpu_time); TEST_ASSERT_EQUAL_FLOAT(84.0, h->mem_usage); } } TEST_ASSERT_TRUE_MESSAGE(live_idx >= 0, "existing host must survive the merge"); TEST_ASSERT_TRUE_MESSAGE(found_new, "new host must be appended by the merge"); /* Cleanup: job_destroy() (jobj.c) releases run_usage via a manual * FREEUP-based walk that (independent of this fix -- a separate, * pre-existing, out-of-scope issue) assumes every host_name is plain * malloc'd. Detach the still-live, still-xstring-backed host_name * before teardown so THAT unrelated issue can't fire inside this * test's own cleanup, and release it correctly ourselves. */ if (live_idx >= 0) got->host_runusage[live_idx]->host_name = NULL; dereference(live_host_name); wru_teardown(); printf( "test_wru_apply_committed_entry_merges_without_freeing_live_host: " "PASS\n"); } /* ── apply_wal_entry(): standby catch-up / replicated-apply dispatch ────── */ static void test_wru_apply_wal_entry_sets_runusage_no_hook(void) { wru_setup(); job_t* j = wru_seed_job(73004, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j); xce_job_runusage_t r = wru_make_test_runusage(); uint8_t* payload = NULL; size_t plen = 0; TEST_ASSERT_EQUAL_INT(0, wru_build_rusage_payload(73004, JOB_STAT_RUN, &r, &payload, &plen)); xce_add_hook(&g_wru_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wru_spy_hook, NULL); int rc = apply_wal_entry(&g_wru_p, WAL_JOB_RUSAGE, 303, payload, (uint32_t)plen); free(payload); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); wru_assert_runusage_matches(j->job.run_usage); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wru_hook_fires, "apply_wal_entry's catch-up dispatch must never fire a " "notification hook"); wru_teardown(); printf("test_wru_apply_wal_entry_sets_runusage_no_hook: PASS\n"); } static void test_wru_apply_wal_entry_backward_compat_old_payload(void) { wru_setup(); job_t* j = wru_seed_job(73005, JOB_STAT_RUN); TEST_ASSERT_NOT_NULL(j); TEST_ASSERT_NULL_MESSAGE(j->job.run_usage, "fixture must seed a job with run_usage still unset"); wal_jobstate_t js = { .jobid = 73005, .state = JOB_STAT_RUN, ._pad = 0 }; int rc = apply_wal_entry(&g_wru_p, WAL_JOB_RUSAGE, 304, &js, sizeof(js)); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); TEST_ASSERT_NULL_MESSAGE(j->job.run_usage, "an old-format WAL_JOB_RUSAGE payload must leave run_usage NULL on " "this seam, unchanged from before this fix"); wru_teardown(); printf("test_wru_apply_wal_entry_backward_compat_old_payload: PASS\n"); } /* ── Restart-replay Observables ──────────────────────────────────────── */ static persist_t* wru_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 wru_teardown_persist(persist_t* p) { persist_stop(p); persist_fini(p); free(p); } /* Encode helper mirroring test_wal_job_start_fact.c's wjs_encode_job * (capnp_encode_job is file-static in persist.c). */ static int wru_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_RUSAGE (real run_usage, same-state RUN->RUN -- the realistic * shape) straight onto disk via persist_wal_async, then replays via a fresh * persist_t's persist_recover() (a real restart). Isolates the WAL fact * shape from the live caller, mirroring test_wal_job_start_fact.c's own * restart-replay test. */ static void test_wru_restart_replay_recovers_runusage(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_wru_replay_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); /* Phase 1: write the log to real disk. */ { persist_t* p = wru_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 = 73006; jdef.state = JOB_STAT_RUN; jdef.submit = &submit; uint8_t* job_payload = NULL; size_t job_plen = 0; TEST_ASSERT_EQUAL_INT(0, wru_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_runusage_t r = wru_make_test_runusage(); uint8_t* rusage_payload = NULL; size_t rusage_plen = 0; TEST_ASSERT_EQUAL_INT(0, wru_build_rusage_payload(73006, JOB_STAT_RUN, &r, &rusage_payload, &rusage_plen)); TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_RUSAGE, rusage_payload, (uint32_t)rusage_plen)); free(rusage_payload); TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); wru_teardown_persist(p); } /* Phase 2: fresh persist_t + fresh cluster_t, recover from disk — * simulates a controller restart. */ { persist_t* p2 = wru_make_fresh_persist(tmp); TEST_ASSERT_NOT_NULL(p2); wru_setup(); p2->cluster = &g_wru_c; int rc = persist_recover(p2); TEST_ASSERT_EQUAL_INT(0, rc); job_t* j = wru_find_job(73006); TEST_ASSERT_NOT_NULL_MESSAGE(j, "restart replay must construct the job via WAL_JOB_SUBMIT"); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); wru_assert_runusage_matches(j->job.run_usage); wru_teardown(); wru_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_wru_restart_replay_recovers_runusage: PASS\n"); } /* Variadic trampoline so the test passes typed args through a real va_list * -- pattern from test_persist_hooks.c / test_commit_barrier_rollout.c's * call_job_hook, reproduced here (file-local) since that one is static to * its own TU. */ static int wru_call_job_hook(int type, int sit, int detail, persist_t* p, ...) { va_list ap; int rc; va_start(ap, p); rc = persist_job_hook_fun(type, sit, detail, p, ap); va_end(ap); return rc; } /* bug-1690's primary regression test: drives the REAL, unmodified * persist_job_hook_fun() -- the actual production hook subscriber this fix * changed -- with a job_t whose run_usage is genuinely populated (see this * file's header comment for why this goes around update_job_rusages() * itself rather than through it), through a real WAL-write + * persist_recover() restart cycle. Asserts run_usage survives the restart * intact, not just state -- the audit's exact confirmed root cause (mirrors * bug-1678/1687/1689's own acceptance criterion). */ static void test_wru_real_hook_appends_and_survives_restart(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_wru_realhook_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); const uint64_t jobid = 73007; /* 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 with run_usage already populated (so the real * persist_job_hook_fun() call site has something to encode) -- mirrors * test_wjs_real_on_job_start_survives_restart's dual-seed pattern. */ { persist_config_t cfg; if (makeString((char*)"wru-realhook-probe") == NULL) xstring_table_init(64); if (!g_wru_jobobj_ready) { TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); g_wru_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); 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)); /* 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_RUN; wal_jdef.submit = &wal_submit; TEST_ASSERT_EQUAL_INT(0, wru_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, with run_usage populated directly * (standing in for update_host_runusage()'s real accumulation -- * see this file's header comment on why that function can't be * exercised for real in this binary), for persist_job_hook_fun() * to find/read. */ job_t* live_job; { 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_RUN; jdef.submit = &submit; live_job = makeJob(c.jobmgr, &jdef); TEST_ASSERT_NOT_NULL(live_job); } xce_job_runusage_t r = wru_make_test_runusage(); live_job->job.run_usage = &r; xce_job_rusages_t rusages; memset(&rusages, 0, sizeof(rusages)); rusages.jobid = jobid; int rc = wru_call_job_hook(XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, HOOK_DETAIL_JOB_RUSAGE_UPDATE, &c.persist, &c, &rusages); TEST_ASSERT_EQUAL_INT(0, rc); /* Prevent the stack-local `r` from being double-freed by teardown * below — this fixture's job_t doesn't own it. */ live_job->job.run_usage = NULL; 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 = wru_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(JOB_STAT_RUN, j2->job.state); /* bug-1690: restart replay must recover the client-reported * run_usage, not just state -- the audit's exact confirmed root * cause (mirrors bug-1678/1687/1689's own acceptance criterion for * their own fields). */ wru_assert_runusage_matches(j2->job.run_usage); xce_hook_fini(&c2.hooksys); if (c2.jobmgr) dereference(c2.jobmgr); wru_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_wru_real_hook_appends_and_survives_restart: PASS\n"); } /* ── Test group entry point ───────────────────────────────────────────── */ void test_wal_job_rusage_fact(void) { UnitySetTestFile(__FILE__); RUN_TEST(test_wru_apply_committed_entry_sets_runusage_no_state_change); RUN_TEST(test_wru_apply_committed_entry_sets_runusage_forward_transition); RUN_TEST(test_wru_apply_committed_entry_backward_compat_old_payload); RUN_TEST(test_wru_apply_committed_entry_merges_without_freeing_live_host); RUN_TEST(test_wru_apply_wal_entry_sets_runusage_no_hook); RUN_TEST(test_wru_apply_wal_entry_backward_compat_old_payload); RUN_TEST(test_wru_restart_replay_recovers_runusage); RUN_TEST(test_wru_real_hook_appends_and_survives_restart); }