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

fix(bug-1731): fire HOOK_DETAIL_JOB_START/JOB_EXIT on WAL apply

Browse source 
apply_committed_entry() (persist.c), the single seam that applies every
committed WAL entry, only ever fired the generic HOOK_DETAIL_JOB_UPDATE
for job-state changes -- HOOK_DETAIL_JOB_START/JOB_EXIT were
permanently dead code, since job.c only ever fires them at
HOOK_BEFORE_EVENT (for WAL-append purposes). This left sched_job_start_
fun/sched_job_exit_fun (sched.c) and apiserver_job_start_fun/
apiserver_job_exit_fun (apiserver.c) unreachable, and the Go apiserver's
already-wired OnJobFinished handler permanently starved of data -- a
finished job's state could stay stuck (e.g. showing RUN) in the
scheduler's own view and the apiserver never learned a job finished.

The wire format was already enriched by bug-1697/bug-1689 with the
pidinfo/exit_info data these hooks need, so this closes the gap rather
than requiring new wire work. WAL_JOB_FINISH now fires
HOOK_DETAIL_JOB_EXIT; WAL_JOB_START now fires HOOK_DETAIL_JOB_START,
gated on a call-scoped decode flag rather than the transition result
(WAL_JOB_START applies as a no-op duplicate for every real job in
production per bug-1697, so gating on the transition result would make
it dead code again). WAL_JOB_FAIL -> HOOK_DETAIL_JOB_FAIL has the same
gap and is left as a follow-up.

Also includes this investigation's supporting changes: a DEBUG3 log-
retention bump (main.c, both controller and scheduler) so race-condition
evidence survives a live burst without rotating out, and DEBUG3-gated
diagnostic trace logging at the two points a job-update notification
can fire or silently drop (job.c, sched.c).

Live-verified end to end on a real deployed cluster: scheduler-side
sched_job_state_change now fires for RUN and EXIT states, and the
apiserver's OnJobFinished -- previously never invoked in production --
now correctly updates job status on completion.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019QxaSzHRB7pmYFp4cVnwkA
This commit is contained in:
Rongsong Shen 2026-08-29 11:29:10 +00:00
parent 42e1eeaa29
commit ebe691c3ed
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
9 changed files with 521 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

182
src/controller/testsuite/test_wal_job_finish_fact.c
View file

@ -80,6 +80,46 @@ static int wjf_spy_hook(int type, int situation, int detail, void* data,
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,
@ -213,9 +253,14 @@ static void test_wjf_apply_committed_entry_sets_finish_time_and_exit_info(void)
(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(1, g_wjf_hook_fires,
"a forward WAL_JOB_FINISH transition must still fire exactly one "
"notification");
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: "
@ -256,6 +301,135 @@ test_wjf_apply_committed_entry_backward_compat_old_payload(void)
"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
@ -682,6 +856,8 @@ 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);