/* test_wal_job_finish_fact.c — bug-1689 regression suite * * Proves WAL_JOB_FINISH's payload -- previously bare wal_jobstate_t{jobid, * state}, exactly the same silent-field-drop bug class as bug-1678's * WAL_JOB_DISPATCH and bug-1687's WAL_JOB_START -- now carries a * receipt-time finish_time AND the worker-reported exit_code (job.exit_info) * 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) * * and that a real, production on_job_finish() call (the actual worker-FINISH * receipt handler) survives a real controller restart (WAL write + * persist_recover() over a real on-disk WAL) with job.exit_info and * job.finish_time (and the derived job.run_time) intact, not just * job.state (Req: mirrors bug-1678/bug-1687's own acceptance criterion, * "after recover the job should get the same as it's state before * shutdown"). * * Also proves backward compatibility: a WAL_JOB_FINISH entry with the OLD * payload shape (exactly sizeof(wal_jobstate_t), no trailing bytes -- * simulating an entry written before this fix) still decodes cleanly, with * exit_info left NULL and finish_time fabricated at apply time exactly as * before this fix (apply_committed_entry) / left untouched exactly as * before this fix (apply_wal_entry) -- no crash, no misread. * * job_exitinfo_t is a single int (exitcode) with NO existing Cap'n Proto * mapping (confirmed -- no JobExitInfo struct in controller.capnp), so the * trailing payload is a small NATIVE (non-capnp) fixed-size trailer, * mirroring wal_jobstate_t's own native-struct style -- unlike bug-1687's * WAL_JOB_START, which needed capnp for the variable-length JobPIDInfo. * * Fixtures mirror test_wal_job_start_fact.c's own conventions exactly: * - wjf_setup/wjf_teardown: bare cluster_t (jobmgr + hooksys) + in-memory * persist_t. * - wjf_make_fresh_persist/wjf_teardown_persist: real disk-backed * persist_t + persist_recover(), for restart-replay tests. */ #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_wjf_jobobj_ready; static cluster_t g_wjf_c; static persist_t g_wjf_p; /* ── Spy hook: counts AFTER_EVENT firings, ignores all payload args ─────── */ static int g_wjf_hook_fires; static int wjf_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_wjf_hook_fires++; return 0; } /* ── bug-1731: detail-aware spy hook, distinguishing HOOK_DETAIL_JOB_EXIT * from HOOK_DETAIL_JOB_UPDATE and capturing the JOB_EXIT payload -- unlike * wjf_spy_hook above (which just counts every AFTER_EVENT fire regardless * of detail), this proves the NEW fire happens (not just "a" fire) and that * the pre-existing JOB_UPDATE fire is not replaced or interfered with. Arg * shapes mirror sched_job_exit_fun/apiserver_job_exit_fun's own va_arg * reads (cluster_t*, xce_jobfinish_t*) and persist_job_hook_fun's * HOOK_DETAIL_JOB_UPDATE case (cluster_t*, job_t*). */ static int g_wjf_exit_hook_fires; static int g_wjf_update_hook_fires; static xce_jobfinish_t g_wjf_captured_finish; static int g_wjf_captured_finish_valid; static int wjf_detail_spy_hook(int type, int situation, int detail, void* data, va_list ap) { (void)type; (void)data; if (situation != HOOK_AFTER_EVENT) return 0; switch (detail) { case HOOK_DETAIL_JOB_EXIT: { (void)va_arg(ap, cluster_t*); xce_jobfinish_t* info = va_arg(ap, xce_jobfinish_t*); g_wjf_exit_hook_fires++; if (info != NULL) { g_wjf_captured_finish = *info; g_wjf_captured_finish_valid = 1; } break; } case HOOK_DETAIL_JOB_UPDATE: g_wjf_update_hook_fires++; break; default: break; } return 0; } static void wjf_setup(void) { kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; if (makeString((char*)"wjf-probe") == NULL) xstring_table_init(64); if (!g_wjf_jobobj_ready) { TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); g_wjf_jobobj_ready = 1; } memset(&g_wjf_c, 0, sizeof(g_wjf_c)); TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&g_wjf_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); g_wjf_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); TEST_ASSERT_NOT_NULL(g_wjf_c.jobmgr); memset(&g_wjf_p, 0, sizeof(g_wjf_p)); g_wjf_p.cluster = &g_wjf_c; g_wjf_hook_fires = 0; } static void wjf_teardown(void) { xce_hook_fini(&g_wjf_c.hooksys); if (g_wjf_c.jobmgr) dereference(g_wjf_c.jobmgr); } /* Look up a job_t wrapper by jobid (NULL if absent). */ static job_t* wjf_find_job(uint64_t jobid) { job_t* j = NULL; lock(g_wjf_c.jobmgr, RTTI_LOCK_READ); int found = key_find(g_wjf_c.jobmgr, (void*)jobid, (void**)&j); unlock(g_wjf_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_start_fact.c's wjs_seed_job. seed_start_time lets FINISH * tests seed a known job.start_time so run_time is derivable/assertable. */ static job_submit_data_t g_wjf_seed_submit; static job_t* wjf_seed_job(uint64_t jobid, uint32_t state, time_t seed_start_time) { xce_job_t jdef; memset(&g_wjf_seed_submit, 0, sizeof(g_wjf_seed_submit)); g_wjf_seed_submit.queue = (char*)"q"; g_wjf_seed_submit.user = (char*)"tester"; memset(&jdef, 0, sizeof(jdef)); jdef.jobid = jobid; jdef.state = state; jdef.start_time = seed_start_time; jdef.submit = &g_wjf_seed_submit; return makeJob(g_wjf_c.jobmgr, &jdef); } /* Build a wal_jobstate_t header + fixed uint64_t finish_time + fixed * int32_t exit_code -- persist_submit_job_finish_fact()'s payload shape. * No capnp needed (job_exitinfo_t is a single int with no existing capnp * mapping), unlike wjs_build_start_payload()'s JobPIDInfo trailer. Caller * owns *out. */ static void wjf_build_finish_payload(uint64_t jobid, uint32_t state, uint64_t finish_time, int32_t exit_code, uint8_t** out, size_t* out_len) { wal_jobstate_t js = { .jobid = jobid, .state = state, ._pad = 0 }; *out_len = sizeof(js) + sizeof(finish_time) + sizeof(exit_code); *out = malloc(*out_len); TEST_ASSERT_NOT_NULL(*out); memcpy(*out, &js, sizeof(js)); memcpy(*out + sizeof(js), &finish_time, sizeof(finish_time)); memcpy(*out + sizeof(js) + sizeof(finish_time), &exit_code, sizeof(exit_code)); } static void wjf_assert_exit_info_matches(job_exitinfo_t* got, int32_t expected_exitcode) { TEST_ASSERT_NOT_NULL_MESSAGE(got, "a WAL_JOB_FINISH payload with a trailing exit_code must set " "job.exit_info (bug-1689), not just state"); TEST_ASSERT_EQUAL_INT(expected_exitcode, got->exitcode); } /* ── apply_committed_entry(): live-commit apply / boot replay dispatch ──── */ static void test_wjf_apply_committed_entry_sets_finish_time_and_exit_info(void) { wjf_setup(); time_t seed_start = 1699999000; job_t* j = wjf_seed_job(73001, JOB_STAT_RUN, seed_start); TEST_ASSERT_NOT_NULL(j); uint64_t wire_finish_time = 1700000050ULL; int32_t wire_exit_code = 17; uint8_t* payload = NULL; size_t plen = 0; wjf_build_finish_payload(73001, JOB_STAT_EXIT, wire_finish_time, wire_exit_code, &payload, &plen); xce_add_hook(&g_wjf_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjf_spy_hook, NULL); int rc = apply_committed_entry(&g_wjf_p, WAL_JOB_FINISH, 300, payload, (uint32_t)plen); free(payload); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_EXIT, j->job.state); TEST_ASSERT_EQUAL_UINT64_MESSAGE(wire_finish_time, (uint64_t)j->job.finish_time, "WAL_JOB_FINISH must adopt the wire-carried finish_time (bug-1689), " "not fabricate time(NULL) when one is present"); wjf_assert_exit_info_matches(j->job.exit_info, wire_exit_code); TEST_ASSERT_EQUAL_INT_MESSAGE( (int)(wire_finish_time - (uint64_t)seed_start), j->job.run_time, "run_time must be re-derived from the wire-carried finish_time minus " "start_time"); TEST_ASSERT_EQUAL_INT_MESSAGE(2, g_wjf_hook_fires, "bug-1731: a forward WAL_JOB_FINISH transition now fires TWO " "notifications, not one -- the pre-existing HOOK_DETAIL_JOB_UPDATE " "plus the new HOOK_DETAIL_JOB_EXIT; wjf_spy_hook counts every " "AFTER_EVENT detail, so both land here. See " "test_wjf_apply_committed_entry_fires_job_exit_hook_exactly_once " "below for a detail-aware assertion isolating the new fire " "specifically"); wjf_teardown(); printf("test_wjf_apply_committed_entry_sets_finish_time_and_exit_info: " "PASS\n"); } static void test_wjf_apply_committed_entry_backward_compat_old_payload(void) { wjf_setup(); time_t seed_start = 1699999000; job_t* j = wjf_seed_job(73002, JOB_STAT_RUN, seed_start); TEST_ASSERT_NOT_NULL(j); TEST_ASSERT_NULL_MESSAGE(j->job.exit_info, "fixture must seed a job with exit_info still unset"); wal_jobstate_t js = { .jobid = 73002, .state = JOB_STAT_DONE, ._pad = 0 }; time_t before = time(NULL); int rc = apply_committed_entry(&g_wjf_p, WAL_JOB_FINISH, 301, &js, sizeof(js)); time_t after = time(NULL); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, j->job.state); TEST_ASSERT_TRUE_MESSAGE( (j->job.finish_time >= before) && (j->job.finish_time <= after), "an old-format (exactly sizeof(wal_jobstate_t)) WAL_JOB_FINISH " "payload must still fabricate finish_time at apply time, unchanged " "from before this fix"); TEST_ASSERT_NULL_MESSAGE(j->job.exit_info, "an old-format WAL_JOB_FINISH payload must leave exit_info NULL -- " "no crash, no misread of absent trailing bytes"); wjf_teardown(); printf("test_wjf_apply_committed_entry_backward_compat_old_payload: " "PASS\n"); } /* bug-1731's primary regression test: apply_committed_entry()'s * WAL_JOB_START/FINISH/... shared case fires exactly ONE generic * HOOK_DETAIL_JOB_UPDATE at HOOK_AFTER_EVENT for every job-state fact, * including WAL_JOB_FINISH -- but never the type-specific * HOOK_DETAIL_JOB_EXIT, so the apiserver's fully-wired `case * NotifyData_data_jobFinishInfo:` (apiserver_job_exit_fun -> OnJobFinished) * was permanently starved of data. This is now fixed: a genuine forward * WAL_JOB_FINISH transition (result == JOB_TRANSITION_APPLIED) also fires * HOOK_DETAIL_JOB_EXIT with a real xce_jobfinish_t{jobid, exit_code} -- RED * before the fix (g_wjf_exit_hook_fires stays 0), GREEN after. Also * confirms the pre-existing HOOK_DETAIL_JOB_UPDATE fire is untouched * (regression guard -- the new fire must not replace or interfere with * it). */ static void test_wjf_apply_committed_entry_fires_job_exit_hook_exactly_once(void) { wjf_setup(); g_wjf_exit_hook_fires = 0; g_wjf_update_hook_fires = 0; g_wjf_captured_finish_valid = 0; memset(&g_wjf_captured_finish, 0, sizeof(g_wjf_captured_finish)); time_t seed_start = 1700100000; job_t* j = wjf_seed_job(73010, JOB_STAT_RUN, seed_start); TEST_ASSERT_NOT_NULL(j); uint64_t wire_finish_time = 1700100050ULL; int32_t wire_exit_code = 7; uint8_t* payload = NULL; size_t plen = 0; wjf_build_finish_payload(73010, JOB_STAT_EXIT, wire_finish_time, wire_exit_code, &payload, &plen); xce_add_hook(&g_wjf_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjf_detail_spy_hook, NULL); int rc = apply_committed_entry(&g_wjf_p, WAL_JOB_FINISH, 400, payload, (uint32_t)plen); free(payload); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_EXIT, j->job.state); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_wjf_exit_hook_fires, "bug-1731: a genuine forward WAL_JOB_FINISH apply must fire " "HOOK_DETAIL_JOB_EXIT at HOOK_AFTER_EVENT exactly once -- this is " "the ONLY mechanism that drives the apiserver's OnJobFinished, " "which was permanently starved of data before this fix (the dead " "handler pair sched_job_exit_fun/apiserver_job_exit_fun only ever " "subscribes to HOOK_AFTER_EVENT, but job.c only ever fires this " "detail at HOOK_BEFORE_EVENT)"); TEST_ASSERT_TRUE_MESSAGE(g_wjf_captured_finish_valid, "HOOK_DETAIL_JOB_EXIT must fire with a non-NULL xce_jobfinish_t*"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(73010, g_wjf_captured_finish.jobid, "the fired xce_jobfinish_t's jobid must match the applied job"); TEST_ASSERT_EQUAL_INT_MESSAGE(wire_exit_code, g_wjf_captured_finish.exit_code, "the fired xce_jobfinish_t's exit_code must match the wire-carried " "exit_code this WAL_JOB_FINISH payload encoded"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_wjf_update_hook_fires, "bug-1731 regression guard: the pre-existing HOOK_DETAIL_JOB_UPDATE " "fire (sched_job_update_fun/apiserver_job_update_fun's live path) " "must still happen too -- the new JOB_EXIT fire must not replace or " "interfere with it"); xce_disable_hook(&g_wjf_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjf_detail_spy_hook); wjf_teardown(); printf( "test_wjf_apply_committed_entry_fires_job_exit_hook_exactly_once: " "PASS\n"); } /* bug-1731 idempotency guard (Constraints: "re-applying an already-applied * WAL_JOB_FINISH should hit the existing result != JOB_TRANSITION_APPLIED * early-return and fire NOTHING, exactly as it does today for * HOOK_DETAIL_JOB_UPDATE"): a job already in a terminal state re-receiving * the identical WAL_JOB_FINISH fact (simulating a restart-replay/standby * catch-up re-application of an already-applied entry) must fire NEITHER * the new HOOK_DETAIL_JOB_EXIT NOR the pre-existing HOOK_DETAIL_JOB_UPDATE * -- the new fire sits inside the SAME `result == JOB_TRANSITION_APPLIED` * guard as JOB_UPDATE, so it inherits this idempotency for free. */ static void test_wjf_apply_committed_entry_reapply_finish_fires_no_hook(void) { wjf_setup(); g_wjf_exit_hook_fires = 0; g_wjf_update_hook_fires = 0; time_t seed_start = 1700200000; /* Already terminal -- as if a prior WAL_JOB_FINISH already applied. */ job_t* j = wjf_seed_job(73011, JOB_STAT_EXIT, seed_start); TEST_ASSERT_NOT_NULL(j); uint64_t wire_finish_time = 1700200050ULL; int32_t wire_exit_code = 3; uint8_t* payload = NULL; size_t plen = 0; wjf_build_finish_payload(73011, JOB_STAT_EXIT, wire_finish_time, wire_exit_code, &payload, &plen); xce_add_hook(&g_wjf_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjf_detail_spy_hook, NULL); int rc = apply_committed_entry(&g_wjf_p, WAL_JOB_FINISH, 401, payload, (uint32_t)plen); free(payload); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wjf_exit_hook_fires, "bug-1731: re-applying an already-terminal WAL_JOB_FINISH " "(rejected-backward/no-op-duplicate) must fire NO " "HOOK_DETAIL_JOB_EXIT -- confirms the new fire inherits the " "existing apply-once idempotency guard for free"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wjf_update_hook_fires, "the pre-existing HOOK_DETAIL_JOB_UPDATE fire must likewise stay " "silent on a duplicate re-apply, unchanged from before this fix"); xce_disable_hook(&g_wjf_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjf_detail_spy_hook); wjf_teardown(); printf( "test_wjf_apply_committed_entry_reapply_finish_fires_no_hook: " "PASS\n"); } /* ── apply_wal_entry(): standby catch-up / replicated-apply dispatch ────── */ static void test_wjf_apply_wal_entry_sets_finish_time_and_exit_info_no_hook(void) { wjf_setup(); time_t seed_start = 1699999100; job_t* j = wjf_seed_job(73003, JOB_STAT_RUN, seed_start); TEST_ASSERT_NOT_NULL(j); uint64_t wire_finish_time = 1700000150ULL; int32_t wire_exit_code = 0; uint8_t* payload = NULL; size_t plen = 0; wjf_build_finish_payload(73003, JOB_STAT_DONE, wire_finish_time, wire_exit_code, &payload, &plen); xce_add_hook(&g_wjf_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjf_spy_hook, NULL); int rc = apply_wal_entry(&g_wjf_p, WAL_JOB_FINISH, 302, payload, (uint32_t)plen); free(payload); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, j->job.state); TEST_ASSERT_EQUAL_UINT64_MESSAGE(wire_finish_time, (uint64_t)j->job.finish_time, "apply_wal_entry's catch-up dispatch must also recover the " "wire-carried finish_time (bug-1689), required for a promoted " "standby"); wjf_assert_exit_info_matches(j->job.exit_info, wire_exit_code); TEST_ASSERT_EQUAL_INT_MESSAGE( (int)(wire_finish_time - (uint64_t)seed_start), j->job.run_time, "apply_wal_entry's catch-up dispatch must also re-derive run_time " "from the wire-carried finish_time"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wjf_hook_fires, "apply_wal_entry's catch-up dispatch must never fire a notification " "hook"); wjf_teardown(); printf("test_wjf_apply_wal_entry_sets_finish_time_and_exit_info_no_hook: " "PASS\n"); } static void test_wjf_apply_wal_entry_backward_compat_old_payload(void) { wjf_setup(); time_t seed_start = 1699999100; job_t* j = wjf_seed_job(73004, JOB_STAT_RUN, seed_start); TEST_ASSERT_NOT_NULL(j); j->job.finish_time = 54321; /* sentinel -- must survive untouched */ j->job.run_time = -1; /* sentinel -- must survive untouched */ wal_jobstate_t js = { .jobid = 73004, .state = JOB_STAT_DONE, ._pad = 0 }; int rc = apply_wal_entry(&g_wjf_p, WAL_JOB_FINISH, 303, &js, sizeof(js)); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_DONE, j->job.state); TEST_ASSERT_EQUAL_UINT64_MESSAGE(54321, (uint64_t)j->job.finish_time, "an old-format WAL_JOB_FINISH payload must leave finish_time " "untouched on this seam, unchanged from before this fix"); TEST_ASSERT_EQUAL_INT_MESSAGE(-1, j->job.run_time, "an old-format WAL_JOB_FINISH payload must leave run_time untouched " "on this seam, unchanged from before this fix"); TEST_ASSERT_NULL_MESSAGE(j->job.exit_info, "an old-format WAL_JOB_FINISH payload must leave exit_info NULL"); wjf_teardown(); printf("test_wjf_apply_wal_entry_backward_compat_old_payload: PASS\n"); } /* ── Restart-replay Observables ──────────────────────────────────────── */ static persist_t* wjf_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 wjf_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 wjf_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 + an old-format * WAL_JOB_START (just to give the job a start_time so run_time is * derivable -- WAL_JOB_START's own richness is bug-1687's concern, not this * file's) + a new-format WAL_JOB_FINISH (real exit_code + a fixed * finish_time) 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_wjf_restart_replay_recovers_finish_time_and_exit_info(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_wjf_replay_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); uint64_t wire_finish_time = 1700019999ULL; int32_t wire_exit_code = 42; /* Phase 1: write the log to real disk. */ { persist_t* p = wjf_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 = 73005; jdef.state = JOB_STAT_PEND; jdef.submit = &submit; uint8_t* job_payload = NULL; size_t job_plen = 0; TEST_ASSERT_EQUAL_INT(0, wjf_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); /* Old-format WAL_JOB_START — just enough to move the job to RUN and * give it SOME start_time (fabricated at apply time); run_time's * correctness is checked below as a derivation from whatever * start_time replay actually produces, not a hardcoded literal. */ wal_jobstate_t start_js = { .jobid = 73005, .state = JOB_STAT_RUN, ._pad = 0 }; TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_START, &start_js, sizeof(start_js))); uint8_t* finish_payload = NULL; size_t finish_plen = 0; wjf_build_finish_payload(73005, JOB_STAT_EXIT, wire_finish_time, wire_exit_code, &finish_payload, &finish_plen); TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_FINISH, finish_payload, (uint32_t)finish_plen)); free(finish_payload); TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); wjf_teardown_persist(p); } /* Phase 2: fresh persist_t + fresh cluster_t, recover from disk — * simulates a controller restart. */ { persist_t* p2 = wjf_make_fresh_persist(tmp); TEST_ASSERT_NOT_NULL(p2); wjf_setup(); p2->cluster = &g_wjf_c; int rc = persist_recover(p2); TEST_ASSERT_EQUAL_INT(0, rc); job_t* j = wjf_find_job(73005); TEST_ASSERT_NOT_NULL_MESSAGE(j, "restart replay must construct the job via WAL_JOB_SUBMIT"); TEST_ASSERT_EQUAL_UINT32(JOB_STAT_EXIT, j->job.state); TEST_ASSERT_EQUAL_UINT64_MESSAGE(wire_finish_time, (uint64_t)j->job.finish_time, "restart replay must recover the exact wire-carried finish_time " "(bug-1689), not a freshly fabricated time(NULL)"); wjf_assert_exit_info_matches(j->job.exit_info, wire_exit_code); TEST_ASSERT_EQUAL_INT_MESSAGE( (int)(wire_finish_time - (uint64_t)j->job.start_time), j->job.run_time, "restart replay must re-derive run_time from the recovered " "finish_time and start_time"); wjf_teardown(); wjf_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_wjf_restart_replay_recovers_finish_time_and_exit_info: " "PASS\n"); } /* bug-1689's primary regression test: drives the REAL production * on_job_finish() path (the actual worker-FINISH receipt handler -- * record_job_finish_fact()'s only caller) with a real xce_jobfinish_t, * through a real WAL-write + persist_recover() restart cycle -- the exact * technique test_wjs_real_on_job_start_survives_restart (bug-1687) already * established. Asserts the job's exit_info AND finish_time survive the * restart intact, not just state -- the audit's exact confirmed root * cause. */ static void test_wjf_real_on_job_finish_survives_restart(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_wjf_onfinish_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); const uint64_t jobid = 73006; const time_t seed_start = 1699999500; const int real_exit_code = 137; time_t before_call, after_call; /* 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 already RUN with a start_time (so the real on_job_finish() call * site has something to find/lock and a meaningful run_time can be * derived) -- mirrors test_wjs_real_on_job_start_survives_restart's * dual-seed pattern exactly. Then drive an actual on_job_finish() call * and let IT append WAL_JOB_FINISH -- this is the production code path * under test, not a hand-built WAL entry. */ { persist_config_t cfg; if (makeString((char*)"wjf-onfinish-probe") == NULL) xstring_table_init(64); if (!g_wjf_jobobj_ready) { TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); g_wjf_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_PEND; wal_jdef.submit = &wal_submit; TEST_ASSERT_EQUAL_INT(0, wjf_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); } /* Durable WAL_JOB_START seed (old-format is fine — bug-1687's own * concern) so replay's own start_time is well-defined. */ { wal_jobstate_t start_js = { .jobid = jobid, .state = JOB_STAT_RUN, ._pad = 0 }; TEST_ASSERT_EQUAL_INT(0, persist_wal_async(&c.persist, WAL_JOB_START, &start_js, sizeof(start_js))); } /* Live in-memory job object, already RUN with a known start_time, * for on_job_finish() 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_RUN; jdef.start_time = seed_start; jdef.submit = &submit; TEST_ASSERT_NOT_NULL(makeJob(c.jobmgr, &jdef)); } /* The real worker-reported finish report. */ xce_jobfinish_t finish = { .jobid = jobid, .exit_code = real_exit_code }; before_call = time(NULL); int rc = on_job_finish(&c, &finish); after_call = time(NULL); TEST_ASSERT_EQUAL_INT(0, rc); 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 = wjf_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_EXIT, j2->job.state, "restart replay must recover the job as EXIT (nonzero exit code) " "from the WAL_JOB_FINISH fact appended by the real " "on_job_finish() call"); /* bug-1689: restart replay must recover not just STATE but the * worker-reported exit_info AND the true finish_time -- the audit's * exact confirmed root cause (mirrors bug-1678/bug-1687's own * acceptance criterion). */ wjf_assert_exit_info_matches(j2->job.exit_info, real_exit_code); TEST_ASSERT_TRUE_MESSAGE( (j2->job.finish_time >= before_call) && (j2->job.finish_time <= after_call), "restart replay must recover the true, receipt-time finish_time " "captured by the real on_job_finish() call, not a fresh " "apply-time time(NULL) fabrication"); TEST_ASSERT_EQUAL_INT_MESSAGE( (int)(j2->job.finish_time - j2->job.start_time), j2->job.run_time, "restart replay must re-derive run_time from the recovered " "finish_time and start_time"); xce_hook_fini(&c2.hooksys); if (c2.jobmgr) dereference(c2.jobmgr); wjf_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_wjf_real_on_job_finish_survives_restart: PASS\n"); } /* ── Test group entry point ───────────────────────────────────────────── */ void test_wal_job_finish_fact(void) { UnitySetTestFile(__FILE__); RUN_TEST(test_wjf_apply_committed_entry_sets_finish_time_and_exit_info); RUN_TEST(test_wjf_apply_committed_entry_backward_compat_old_payload); RUN_TEST(test_wjf_apply_committed_entry_fires_job_exit_hook_exactly_once); RUN_TEST(test_wjf_apply_committed_entry_reapply_finish_fires_no_hook); RUN_TEST(test_wjf_apply_wal_entry_sets_finish_time_and_exit_info_no_hook); RUN_TEST(test_wjf_apply_wal_entry_backward_compat_old_payload); RUN_TEST(test_wjf_restart_replay_recovers_finish_time_and_exit_info); RUN_TEST(test_wjf_real_on_job_finish_survives_restart); }