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

fix(controller): WAL_JOB_FINISH loses exit_info/finish_time on replay -- and exit_info was never set at all in production (bug-1689)

Browse source 
Same class as bug-1678/bug-1687: record_job_finish_fact() wrote only
{jobid, state} for WAL_JOB_FINISH. Both apply seams either fabricated
finish_time via time(NULL) or touched nothing beyond state.

Worse than "lost on restart": apply_job_finish_transition() -- the only
function that ever set job.exit_info -- is dead code on the real
production apply path. The raft-apply-decoupling redesign routes actual
mutation through the leaner apply_job_update_transition() instead, so
exit_info was never populated at all, restart or not, before this fix.

job_exitinfo_t ({int exitcode;}) has no existing capnp mapping, so this
uses a small native (non-capnp) trailing payload -- wal_jobstate_t header
+ uint64_t finish_time + int32_t exit_code -- mirroring wal_jobstate_t's
own native-struct style rather than inventing a capnp wrapper for one
field. run_time is never carried on the wire, always re-derived as
finish_time - start_time at apply time. Both apply seams updated;
old-format payloads (exactly sizeof(wal_jobstate_t)) decode unchanged
(fabricated finish_time, NULL exit_info) -- full backward compatibility.

Found by an /xce-audit-invariants controller pass (field-complete apply
parity dimension). 6 new tests in test_wal_job_finish_fact.c; RED/GREEN
independently verified (4 genuine pre-fix failures with real
expected-vs-fabricated mismatches, 2 backward-compat tests correctly
passed both before and after). Full controller suite: 1152 tests, 2
failures -- both the pre-existing, unrelated bug-1561 overlayfs sandbox
flake (confirmed flaky across repeated re-runs of the same binary).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AgYRU3zDTqNGV5Jk8aiKLg
This commit is contained in:
Rongsong Shen 2026-08-25 09:58:55 +00:00
parent 44bc50202a
commit c2690cbf95
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
8 changed files with 1381 additions and 425 deletions
Download patch file Download diff file Expand all files Collapse all files

865
.wolf/buglog.json
View file

File diff suppressed because one or more lines are too long

65
src/controller/job.c
View file

@ -851,6 +851,45 @@ int restore_job_start_state(cluster_t* c, uint64_t jobid, uint32_t state,
return 0;
}
/* bug-1689: WAL_JOB_FINISH's own sibling of restore_job_start_state()/
* restore_job_dispatch_state() above (bug-1687/bug-1678), for
* apply_wal_entry()'s standby catch-up / replicated-apply seam -- takes
* ownership of an optional decoded job_exitinfo_t (NULL for a state-only
* WAL_JOB_FINISH entry, e.g. one written before this fix) and a
* wire-carried finish_time (0 meaning "not carried -- leave
* job.finish_time/run_time untouched", matching this seam's own pre-fix
* behavior of never touching them for WAL_JOB_FINISH at all) so a promoted
* standby recovers the worker-confirmed exit_code and the true finish_time
* the same way apply_committed_entry's WAL_JOB_FINISH case already does for
* the primary's own boot replay. job_exitinfo_t has no nested pointers (a
* single int), so it is released with plain FREEUP, not a capnp
* free_*_ptr helper. */
int restore_job_finish_state(cluster_t* c, uint64_t jobid, uint32_t state,
uint64_t finish_time, job_exitinfo_t* exit_info)
{
job_t* j = NULL;
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)jobid, (void**)&j);
unlock(c->jobmgr);
if (!j) {
FREEUP(exit_info);
return -ENOENT;
}
lock(j, RTTI_LOCK_WRITE);
j->job.state = (int)state;
if (finish_time != 0)
j->job.finish_time = (time_t)finish_time;
if (exit_info != NULL) {
FREEUP(j->job.exit_info);
j->job.exit_info = exit_info;
}
if (finish_time != 0)
j->job.run_time = (int)(j->job.finish_time - j->job.start_time);
unlock(j);
return 0;
}
static int get_nslots_in_alloc(xce_host_alloc_t* alloc)
{
int i;
@ -2349,10 +2388,28 @@ static int record_job_finish_fact(cluster_t* c, job_t* job,
/* Durable append only — no mutation, no AFTER_EVENT firing here (Req
* 5.1, 5.3). The resultant state is deterministic from finish->exit_code
* alone, independent of current job state. */
(void)persist_submit_job_fact(&c->persist, WAL_JOB_FINISH, job->job.jobid,
(finish->exit_code == 0) ? JOB_STAT_DONE
: JOB_STAT_EXIT);
* alone, independent of current job state.
*
* bug-1689: persist_submit_job_finish_fact() carries a receipt-time
* finish_time and the worker-reported exit_code alongside {jobid,
* state} -- see that function's doc comment and wal_types.h's
* WAL_JOB_FINISH payload note for the full rationale (mirrors
* bug-1687's WAL_JOB_START fix). Before this fix, persist_submit_job_
* fact()'s bare {jobid, state} payload silently dropped both: job.
* exit_info replayed permanently NULL and job.finish_time (and its
* derived job.run_time) were re-fabricated (time(NULL)) fresh at every
* apply instead of carrying the true receipt-time value. Capturing
* finish_time HERE, once, at receipt time (rather than re-deriving
* time(NULL) independently at each of apply_committed_entry's/
* apply_wal_entry's OWN apply-time moments) makes it the single
* authoritative value both apply seams converge on. */
time_t finish_time = time(NULL);
(void)persist_submit_job_finish_fact(&c->persist, job->job.jobid,
(finish->exit_code == 0)
? JOB_STAT_DONE
: JOB_STAT_EXIT,
(uint64_t)finish_time,
finish->exit_code);
return 0;
}

1
src/controller/meson.build
View file

@ -171,6 +171,7 @@ controller_unit_test_src = [
'testsuite/test_replay_apply_seam.c',
'testsuite/test_wal_job_dispatch_fact.c',
'testsuite/test_wal_job_start_fact.c',
'testsuite/test_wal_job_finish_fact.c',
'testsuite/test_job_dispatch_fail_count.c',
'testsuite/test_apply_reapply_idempotency.c',
'testsuite/test_apply_correctness_verification.c',

145
src/controller/persist.c
View file

@ -52,6 +52,17 @@ extern int restore_job_dispatch_state(cluster_t* c, uint64_t jobid,
extern int restore_job_start_state(cluster_t* c, uint64_t jobid,
uint32_t state, uint64_t start_time,
job_pidinfo_t* pidinfo);
/* restore_job_finish_state defined in job.c (bug-1689): WAL_JOB_FINISH's own
* sibling of restore_job_state() above, additionally taking ownership of an
* optional decoded job_exitinfo_t and a wire-carried finish_time (0 means
* "not carried -- leave job.finish_time/run_time untouched", the pre-fix
* payload's behavior) so the worker-confirmed finish survives standby
* catch-up / replicated-apply the same way apply_committed_entry's
* WAL_JOB_FINISH case already restores it on the primary's own boot
* replay. */
extern int restore_job_finish_state(cluster_t* c, uint64_t jobid,
uint32_t state, uint64_t finish_time,
job_exitinfo_t* exit_info);
extern int worker_from_capnp(cluster_t* c,
const void* capnp_payload, size_t len);
extern int restore_queue_state(cluster_t* c,
@ -264,6 +275,31 @@ static job_pidinfo_t* decode_pidinfo(const uint8_t* payload, size_t len)
return info;
}
/* Decode a native (non-capnp) WAL_JOB_FINISH exit_code trailer (bug-1689)
* into a freshly heap-allocated job_exitinfo_t*. job_exitinfo_t (xce/types.h)
* is a single int with NO existing Cap'n Proto mapping (confirmed -- no
* JobExitInfo struct in controller.capnp), so this mirrors wal_jobstate_t's
* own native-struct style rather than inventing a capnp wrapper for one int
* -- unlike decode_joballoc()/decode_pidinfo() above, which decode genuine
* capnp payloads. Returns NULL on a short/absent payload; caller owns the
* result (release via free()/FREEUP -- no nested pointers). */
static job_exitinfo_t* decode_jobexitinfo(const uint8_t* payload, size_t len)
{
job_exitinfo_t* info;
int32_t exitcode;
if (!payload || len < sizeof(exitcode))
return NULL;
memcpy(&exitcode, payload, sizeof(exitcode));
info = (job_exitinfo_t*)calloc(1, sizeof(*info));
if (!info)
return NULL;
info->exitcode = (int)exitcode;
return info;
}
/* Serialize a host admin-state frame (config-upgrade-snapshot task 2.1, Req
* 1.6/1.7/1.9) to a malloc'd Cap'n Proto buffer. Mirrors capnp_encode_job
* above exactly -- capnp_host_admin_state_t (typemap.h) is a frame-only
@ -3579,7 +3615,6 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
restore_job_array(p->cluster, payload, payload_len, lsn, NULL);
break;
case WAL_JOB_FINISH:
case WAL_JOB_FAIL:
case WAL_JOB_STOP:
case WAL_JOB_RESUME:
@ -3631,6 +3666,44 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
break;
}
case WAL_JOB_FINISH: {
/* bug-1689: split out of the shared group above -- mirrors
* WAL_JOB_START's split just above (bug-1687) / WAL_JOB_DISPATCH's
* split just below (bug-1678). This is the one fact type in that
* group whose payload can carry a fixed finish_time PLUS a fixed
* exit_code (persist_submit_job_finish_fact's shape) alongside the
* {jobid, state} header, so a promoted standby's catch-up replay
* recovers the worker-confirmed exit_code and the true finish_time
* too, not just state. A payload of exactly sizeof(wal_jobstate_t)
* (no trailing bytes -- any WAL_JOB_FINISH entry written before this
* fix) decodes neither, matching the pre-fix state-only behavior
* exactly (finish_time/run_time/exit_info all left untouched on
* this seam). */
if (payload_len >= sizeof(wal_jobstate_t)) {
const wal_jobstate_t* js = (const wal_jobstate_t*)payload;
uint64_t finish_time = 0;
job_exitinfo_t* exit_info = NULL;
uint32_t extra = payload_len - (uint32_t)sizeof(wal_jobstate_t);
if (extra >= sizeof(finish_time)) {
memcpy(&finish_time,
(const uint8_t*)payload + sizeof(wal_jobstate_t),
sizeof(finish_time));
uint32_t exitinfo_len = extra - (uint32_t)sizeof(finish_time);
if (exitinfo_len > 0) {
exit_info = decode_jobexitinfo(
(const uint8_t*)payload + sizeof(wal_jobstate_t)
+ sizeof(finish_time),
exitinfo_len);
}
}
restore_job_finish_state(p->cluster, js->jobid, js->state,
finish_time, exit_info);
}
break;
}
case WAL_JOB_DISPATCH: {
/* bug-1678: split out of the shared group above -- this is the one
* fact type whose payload can carry a trailing Cap'n Proto JobAlloc
@ -3973,7 +4046,42 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn,
j->job.start_time = time(NULL);
}
} else if (wal_type == WAL_JOB_FINISH) {
j->job.finish_time = time(NULL);
/* bug-1689: a WAL_JOB_FINISH payload can carry a fixed
* finish_time plus a fixed exit_code beyond the
* wal_jobstate_t header (persist_submit_job_finish_fact's
* shape, mirroring bug-1687's WAL_JOB_START trailing-fields
* handling above). Decode and adopt the wire-carried
* finish_time/exit_code when present; a payload of exactly
* sizeof(wal_jobstate_t) (no trailing bytes -- any
* WAL_JOB_FINISH entry written before this fix)
* intentionally falls through to the old fabricated-
* time(NULL)/exit_info-untouched behavior for full backward
* compatibility. run_time is always re-derived from
* finish_time - start_time either way. */
if (len >= sizeof(wal_jobstate_t) + sizeof(uint64_t)) {
uint64_t wire_finish_time;
memcpy(&wire_finish_time,
(const uint8_t*)payload + sizeof(wal_jobstate_t),
sizeof(wire_finish_time));
j->job.finish_time = (time_t)wire_finish_time;
uint32_t exitinfo_len = len
- (uint32_t)sizeof(wal_jobstate_t)
- (uint32_t)sizeof(uint64_t);
if (exitinfo_len > 0) {
job_exitinfo_t* decoded = decode_jobexitinfo(
(const uint8_t*)payload + sizeof(wal_jobstate_t)
+ sizeof(uint64_t),
exitinfo_len);
if (decoded != NULL) {
if (j->job.exit_info != NULL)
free(j->job.exit_info);
j->job.exit_info = decoded;
}
}
} else {
j->job.finish_time = time(NULL);
}
j->job.run_time = (int)(j->job.finish_time - j->job.start_time);
} else if (wal_type == WAL_JOB_DISPATCH
&& len > sizeof(wal_jobstate_t)) {
@ -7034,6 +7142,39 @@ int persist_submit_job_start_fact(persist_t* p, uint64_t jobid, int state,
return rc;
}
/* bug-1689: WAL_JOB_FINISH's own richer sibling of persist_submit_job_fact()
* -- mirrors persist_submit_job_start_fact() (bug-1687) exactly, but for the
* worker-confirmed-finish fact: the sole durable record of the job's real
* exit_code and its true finish_time. Payload shape: wal_jobstate_t header,
* immediately followed by a fixed uint64_t finish_time (the receipt-time
* clock, captured ONCE by the caller rather than re-derived at each apply
* seam's own apply-time moment), immediately followed by a fixed int32_t
* exit_code -- a small NATIVE trailer, not capnp (job_exitinfo_t is a single
* int with no existing capnp mapping, unlike WAL_JOB_START's variable-length
* JobPIDInfo, so no malloc is needed here -- the whole payload is a fixed
* size). apply_committed_entry()'s and apply_wal_entry()'s WAL_JOB_FINISH
* cases decode exactly this shape, and treat a payload of exactly
* sizeof(wal_jobstate_t) (no trailing bytes) as "state-only, fabricate
* finish_time and leave exit_info untouched" for backward compatibility
* with any WAL_JOB_FINISH entry written before this fix. run_time is NOT
* carried -- it is re-derived (finish_time - start_time) at each apply seam
* once both are known. */
int persist_submit_job_finish_fact(persist_t* p, uint64_t jobid, int state,
uint64_t finish_time, int exit_code)
{
wal_jobstate_t js = { .jobid = jobid, .state = (uint32_t)state,
._pad = 0 };
int32_t ec = (int32_t)exit_code;
uint8_t payload[sizeof(js) + sizeof(finish_time) + sizeof(ec)];
memcpy(payload, &js, sizeof(js));
memcpy(payload + sizeof(js), &finish_time, sizeof(finish_time));
memcpy(payload + sizeof(js) + sizeof(finish_time), &ec, sizeof(ec));
return wal_hook_append_classified(p, WAL_JOB_FINISH, WAL_MODE_RETAIN,
payload, (uint32_t)sizeof(payload));
}
/* bug-1679: see persist.h's doc comment for the full rationale. Mirrors
* apply_committed_entry()'s own identical Hazard-1-guarded hook fire
* (its shared WAL_JOB_START/.../DISPATCH case, right after its

2
src/controller/testsuite/controller_test.c
View file

@ -85,6 +85,7 @@ extern void test_apply_committed_entry(void);
extern void test_replay_apply_seam(void);
extern void test_wal_job_dispatch_fact(void);
extern void test_wal_job_start_fact(void);
extern void test_wal_job_finish_fact(void);
extern void test_job_dispatch_fail_count(void);
extern void test_apply_reapply_idempotency(void);
extern void test_apply_correctness_verification(void);
@ -256,6 +257,7 @@ int main(int argc, char* argv[])
TEST(test_replay_apply_seam);
TEST(test_wal_job_dispatch_fact);
TEST(test_wal_job_start_fact);
TEST(test_wal_job_finish_fact);
TEST(test_job_dispatch_fail_count);
TEST(test_apply_reapply_idempotency);
TEST(test_apply_correctness_verification);

689
src/controller/testsuite/test_wal_job_finish_fact.c Normal file
View file

@ -0,0 +1,689 @@
/* 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 <rtti.h>
#include <capnp_c.h>
#include <errno.h>
#include <stdarg.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#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;
}
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(1, g_wjf_hook_fires,
"a forward WAL_JOB_FINISH transition must still fire exactly one "
"notification");
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");
}
/* ── 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_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);
}

20
src/include/persist.h
View file

@ -1437,6 +1437,26 @@ int persist_submit_job_start_fact(persist_t* p, uint64_t jobid, int state,
uint64_t start_time,
const job_pidinfo_t* info);
/* bug-1689: WAL_JOB_FINISH's own sibling of persist_submit_job_fact(),
* mirroring persist_submit_job_start_fact() (bug-1687) exactly but for the
* worker-confirmed-finish fact -- additionally encoding a receipt-time
* finish_time AND the worker-reported exit_code into the fact's payload
* (header {jobid, state} + a fixed uint64_t finish_time + a fixed int32_t
* exit_code) so a restart/catch-up replay can reconstruct job.exit_info and
* the true job.finish_time instead of leaving exit_info permanently NULL and
* finish_time fabricated fresh (time(NULL)) at every apply. job_exitinfo_t
* is a single int with no existing Cap'n Proto mapping (no JobExitInfo
* struct in controller.capnp) -- this is a small NATIVE trailing payload,
* mirroring wal_jobstate_t's own native-struct style, not a capnp wrapper
* (unlike persist_submit_job_start_fact()'s variable-length JobPIDInfo,
* which genuinely needed one). run_time is NOT carried on the wire -- it is
* a pure derivation (finish_time - start_time), re-derived at each apply
* seam once both are known, mirroring apply_job_finish_transition()'s own
* formula. Returns the local-append rc, same contract as
* persist_submit_job_fact(). */
int persist_submit_job_finish_fact(persist_t* p, uint64_t jobid, int state,
uint64_t finish_time, int exit_code);
/* bug-1679: fire XCE_CONTROLLER_HOOK_JOB's HOOK_DETAIL_JOB_UPDATE AFTER_EVENT
* for a job whose durable WAL fact the caller has ALREADY appended itself
* (e.g. dispatch_a_job_()'s own persist_submit_job_dispatch_fact() call,

19
src/include/wal_types.h
View file

@ -112,7 +112,24 @@ typedef enum {
* seams treat a payload of exactly
* sizeof(wal_jobstate_t) as the legacy state-only
* shape for full backward compatibility. */
WAL_JOB_FINISH = 3,
WAL_JOB_FINISH = 3, /* worker-confirmed job finish. Payload is
* wal_jobstate_t{jobid, state=DONE|EXIT, _pad=0}
* for any entry written before bug-1689; as of
* bug-1689 the real producer
* (persist_submit_job_finish_fact) appends a fixed
* uint64_t finish_time (receipt-time clock, not
* fabricated at apply time) plus a fixed int32_t
* exit_code -- both apply seams treat a payload of
* exactly sizeof(wal_jobstate_t) as the legacy
* state-only shape for full backward
* compatibility. job_exitinfo_t is a single int
* with no existing Cap'n Proto mapping, so this is
* a small NATIVE trailing payload (mirroring
* wal_jobstate_t's own native-struct style), not a
* capnp wrapper. run_time is NOT carried on the
* wire -- it is a pure derivation (finish_time -
* start_time), re-derived at each apply seam once
* both are known. */
WAL_JOB_FAIL = 4,
WAL_JOB_STOP = 5,
WAL_JOB_RESUME = 6,