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

fix(controller): WAL_JOB_RUSAGE loses job.run_usage on replay (bug-1690)

Browse source 
Same class as bug-1678/1687/1689: the WAL fact for WAL_JOB_RUSAGE only
carried {jobid, state}, so job.run_usage (client-reported per-host
CPU/mem usage) was never restored on either apply seam after a restart.

Unlike the prior three, xce_job_runusage_t/xce_host_runusage_t had no
existing Cap'n Proto mapping -- a genuine new schema addition
(HostRunusage/JobRunusage in controller.capnp, mirroring JobPendReason's
List(NestedStruct)+$C.maplistcount style), verified against the real
struct definitions in types.h, codegen confirmed via nm on the built
binary rather than assumed.

Two deliberate deviations from the established precedent, found by
reading the real code rather than copying the pattern blindly:

1. Adoption is unconditional, not gated on a state transition -- a
   rusage report almost never changes job.state (repeat reports keep
   state at RUN), so gating on JOB_TRANSITION_APPLIED like START/FINISH/
   DISPATCH do would make the fix dead code for the realistic case.

2. Merge, not free-and-replace. update_host_runusage() builds run_usage
   with host_name sourced from makeString() -- an RTTI-managed pointer,
   not malloc() -- and the primary's own live apply pump re-applies every
   fact it just wrote. A naive free_JobRunusage_ptr()-then-adopt (matching
   the prior three fixes exactly) would free() a non-malloc'd pointer on
   the very first live rusage report for a new host -- guaranteed heap
   corruption. Fixed with merge_job_runusage(): existing hosts get
   scalar-only in-place updates (host_name pointer never touched), only
   genuinely new hosts adopt their freshly-decoded (safe, malloc'd)
   host_name. Proved via a real makeString()-backed test asserting
   pointer identity survives, not just value equality. job_destroy()
   (src/lib/jobj.c) was found to have this identical latent mismatch,
   pre-existing and independent of this fix -- flagged, not fixed here.

Found by an /xce-audit-invariants controller pass (field-complete apply
parity dimension). 8 new tests in test_wal_job_rusage_fact.c; RED/GREEN
independently verified. Full controller suite: 1160 tests, 2 failures --
both the pre-existing, unrelated bug-1561 overlayfs sandbox flake. No
ASAN/heap-corruption findings. Backward compatibility confirmed for
old-format bare-payload entries.

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 10:53:32 +00:00
parent c3d6cfa4cc
commit f7b732a300
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
8 changed files with 1328 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

31
.wolf/buglog.json
View file

File diff suppressed because one or more lines are too long

133
src/controller/job.c
View file

@ -890,6 +890,139 @@ int restore_job_finish_state(cluster_t* c, uint64_t jobid, uint32_t state,
return 0;
}
/* bug-1690: merge (never wholesale free-and-replace) a freshly-decoded
* xce_job_runusage_t into *dst. Takes ownership of `src` -- always
* consumed/freed by this call (either adopted whole, partially cannibalized
* for its new-host entries, or fully released), regardless of outcome.
*
* WHY A MERGE, NOT A REPLACE: *dst can be the job's LIVE run_usage --
* update_job_rusages() (job.c) mutates it synchronously and in place, via
* update_host_runusage() (misc.c), BEFORE the corresponding WAL fact is even
* appended. On the primary, persist_apply_pump_run() (persist.c) later
* re-applies that SAME fact via apply_committed_entry() moments afterward,
* on the SAME live job object -- so *dst can genuinely be that live,
* already-current structure at the moment this runs, not just a stale
* replay artifact. Critically, update_host_runusage() sources each
* xce_host_runusage_t.host_name from makeString() (src/lib/xstring.c) -- an
* RTTI-managed, refcounted intern-table pointer, NOT a plain malloc()
* allocation. RTTI's own header-before-data layout (librtti's
* RTTI_DATA_TO_OBJ: the object header sits BEFORE the pointer callers hold)
* means a plain free() on that pointer -- which the capnp-generated
* free_JobRunusage_ptr()/free_HostRunusage_ptr() would do, being written for
* genuinely capnp-decoded, plain-malloc'd Text fields -- corrupts the heap.
* (job_destroy()'s own manual FREEUP-based walk over run_usage, jobj.c, has
* this exact same latent mismatch already, independent of this fix -- flagged
* separately, not fixed here, since jobj.c/misc.c are outside this fix's
* scope.)
*
* The merge below NEVER frees an existing host_name: an already-present
* host (matched by name) only has its scalar cpu_time/mem_usage overwritten
* in place; only a genuinely NEW host's own freshly-decoded (always
* plain-malloc'd) host_name is adopted, by ownership transfer, never a
* live/xstring one. `src`'s own shell plus every entry NOT transferred is
* released via the ordinary capnp free path at the end -- safe, since `src`
* is always capnp-decoded, end to end, regardless of what *dst was. */
void merge_job_runusage(xce_job_runusage_t** dst, xce_job_runusage_t* src)
{
xce_job_runusage_t* d;
int i, k;
if (*dst == NULL) {
/* Nothing live to preserve — adopt the decoded copy outright,
* exactly like every other restore_job_*_state() sibling. */
*dst = src;
return;
}
d = *dst;
/* Scalar job-level fields: no pointers, always safe to overwrite in
* place regardless of *dst's provenance. */
d->max_mem = src->max_mem;
d->avg_mem = src->avg_mem;
d->used_mem = src->used_mem;
d->cpu_time = src->cpu_time;
d->peak_cpu_time = src->peak_cpu_time;
d->peak_cpu_duration = src->peak_cpu_duration;
d->mem_efficiency = src->mem_efficiency;
d->avg_cpu_efficiency = src->avg_cpu_efficiency;
d->peak_cpu_efficiency = src->peak_cpu_efficiency;
for (i = 0; i < src->num_hosts; i++) {
xce_host_runusage_t* sh = src->host_runusage[i];
int matched = -1;
if (sh == NULL)
continue;
for (k = 0; k < d->num_hosts; k++) {
if (d->host_runusage[k] != NULL &&
d->host_runusage[k]->host_name != NULL &&
sh->host_name != NULL &&
strcmp(sh->host_name, d->host_runusage[k]->host_name) == 0) {
matched = k;
break;
}
}
if (matched >= 0) {
/* Existing host: overwrite scalars only. d's host_name pointer
* (possibly xstring-backed) is never touched or freed here. */
d->host_runusage[matched]->cpu_time = sh->cpu_time;
d->host_runusage[matched]->mem_usage = sh->mem_usage;
} else {
/* Genuinely new host: append, transferring ownership of sh's
* own plain-malloc'd host_name directly -- never xstring. */
xce_host_runusage_t** grown = (xce_host_runusage_t**)realloc(
d->host_runusage,
(size_t)(d->num_hosts + 1) * sizeof(xce_host_runusage_t*));
if (grown == NULL)
continue; /* OOM -- skip this host, keep the rest merged */
d->host_runusage = grown;
d->host_runusage[d->num_hosts] = sh;
d->num_hosts++;
src->host_runusage[i] = NULL; /* ownership transferred -- the
* free_JobRunusage_ptr() call
* below must not also free it */
}
}
/* Release src's shell + every entry whose ownership was NOT transferred
* above -- always safe, src is capnp-decoded end to end. */
free_JobRunusage_ptr(&src);
}
/* bug-1690: WAL_JOB_RUSAGE's own sibling of restore_job_state() above, for
* apply_wal_entry()'s standby catch-up / replicated-apply seam -- takes
* ownership of an optional decoded xce_job_runusage_t (NULL for a
* state-only WAL_JOB_RUSAGE entry -- either one written before this fix, or
* one fired while job.run_usage was still NULL at write time) so a promoted
* standby recovers the client-reported per-host CPU/mem usage the same way
* apply_committed_entry's WAL_JOB_RUSAGE case already does for the primary's
* own boot replay. Unconditional absolute state overwrite, matching every
* other restore_job_*_state() sibling on this seam -- no transition-result
* guarding happens on standby catch-up. Merges (never frees-and-replaces)
* run_usage -- see merge_job_runusage()'s doc comment above for why. */
int restore_job_rusage_state(cluster_t* c, uint64_t jobid, uint32_t state,
xce_job_runusage_t* runusage)
{
job_t* j = NULL;
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)jobid, (void**)&j);
unlock(c->jobmgr);
if (!j) {
if (runusage != NULL)
free_JobRunusage_ptr(&runusage);
return -ENOENT;
}
lock(j, RTTI_LOCK_WRITE);
j->job.state = (int)state;
if (runusage != NULL)
merge_job_runusage(&(j->job.run_usage), runusage);
unlock(j);
return 0;
}
static int get_nslots_in_alloc(xce_host_alloc_t* alloc)
{
int i;

1
src/controller/meson.build
View file

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

191
src/controller/persist.c
View file

@ -63,6 +63,30 @@ extern int restore_job_start_state(cluster_t* c, uint64_t jobid,
extern int restore_job_finish_state(cluster_t* c, uint64_t jobid,
uint32_t state, uint64_t finish_time,
job_exitinfo_t* exit_info);
/* restore_job_rusage_state defined in job.c (bug-1690): WAL_JOB_RUSAGE's own
* sibling of restore_job_state() above, additionally taking ownership of an
* optional decoded xce_job_runusage_t (NULL for a state-only WAL_JOB_RUSAGE
* entry -- either one written before this fix, or one fired while
* job.run_usage was still NULL) so the client-reported per-host CPU/mem
* usage survives standby catch-up / replicated-apply the same way
* apply_committed_entry's WAL_JOB_RUSAGE case already restores it on the
* primary's own boot replay. */
extern int restore_job_rusage_state(cluster_t* c, uint64_t jobid,
uint32_t state,
xce_job_runusage_t* runusage);
/* merge_job_runusage defined in job.c (bug-1690): merges a freshly-decoded
* xce_job_runusage_t into *dst rather than freeing-and-replacing it. Needed
* here too (apply_committed_entry's own WAL_JOB_RUSAGE branch below), not
* just by restore_job_rusage_state() above -- *dst can be the job's LIVE
* run_usage (update_job_rusages()/update_host_runusage(), misc.c, mutate it
* synchronously and in place, with an xstring-backed host_name, BEFORE the
* WAL fact is even appended), and persist_apply_pump_run() re-applies that
* SAME fact on the SAME live job moments later via apply_committed_entry --
* a plain free()-and-replace (e.g. the capnp-generated free_JobRunusage_ptr)
* would free a non-malloc'd, RTTI-managed pointer there. See
* merge_job_runusage()'s own doc comment (job.c) for the full rationale. */
extern void merge_job_runusage(xce_job_runusage_t** dst,
xce_job_runusage_t* src);
extern int worker_from_capnp(cluster_t* c,
const void* capnp_payload, size_t len);
extern int restore_queue_state(cluster_t* c,
@ -300,6 +324,66 @@ static job_exitinfo_t* decode_jobexitinfo(const uint8_t* payload, size_t len)
return info;
}
/* Serialize xce_job_runusage_t to a malloc'd Cap'n Proto buffer (bug-1690):
* lets WAL_JOB_RUSAGE carry the client-reported, accumulated per-host
* CPU/mem usage (job->job.run_usage) alongside the {jobid, state} header, so
* a restart/catch-up replay can reconstruct it instead of leaving it
* permanently NULL. Mirrors capnp_encode_joballoc()/capnp_encode_pidinfo()
* above exactly, using the NEW JobRunusage struct (controller.capnp) --
* unlike JobAlloc/JobPIDInfo, xce_job_runusage_t had NO existing Cap'n Proto
* mapping at all, so this is a genuine schema addition (see the schema's own
* doc comment on JobRunusage for why it is not the same type as
* xce_job_rusages_t/JobRusages, a similar-looking but unrelated struct). */
static int capnp_encode_jobrunusage(const 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;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_JobRunusage(cs, &cjr, (xce_job_runusage_t*)r);
jrp = new_JobRunusage(cs);
write_JobRunusage(&cjr, jrp);
capn_setp(capn_root(&c), 0, jrp.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;
}
/* Decode a Cap'n Proto JobRunusage buffer (bug-1690) into a freshly heap-
* allocated xce_job_runusage_t*, mirroring decode_joballoc()/decode_pidinfo()'s
* pattern above. Returns NULL on a NULL/empty payload; caller owns the
* result (release via free_JobRunusage_ptr). */
static xce_job_runusage_t* decode_jobrunusage(const uint8_t* payload,
size_t len)
{
struct capn c_ctx;
JobRunusage_ptr jrp;
struct JobRunusage cjr;
xce_job_runusage_t* r;
if (!payload || len == 0)
return NULL;
capn_init_mem(&c_ctx, payload, len, 0);
jrp.p = capn_getp(capn_root(&c_ctx), 0, 1);
memset(&cjr, 0, sizeof(cjr));
read_JobRunusage(&cjr, jrp);
r = calloc(1, sizeof(*r));
if (!r) { capn_free(&c_ctx); return NULL; }
decode_JobRunusage(r, &cjr);
capn_free(&c_ctx);
return r;
}
/* 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
@ -3621,7 +3705,6 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
case WAL_JOB_UPDATE:
case WAL_JOB_CKPT:
case WAL_JOB_PREASON:
case WAL_JOB_RUSAGE:
/* Payload: wal_jobstate_t {jobid, state} */
if (payload_len >= sizeof(wal_jobstate_t)) {
const wal_jobstate_t* js = (const wal_jobstate_t*)payload;
@ -3629,6 +3712,34 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
}
break;
case WAL_JOB_RUSAGE: {
/* bug-1690: split out of the shared group above -- mirrors
* WAL_JOB_START's/WAL_JOB_FINISH's splits just below (bug-1687/
* bug-1689) and WAL_JOB_DISPATCH's split further below (bug-1678).
* This is the one fact type in that group whose payload can carry
* trailing Cap'n Proto JobRunusage bytes (persist_job_hook_fun's
* WAL_JOB_RUSAGE payload shape, appended only when job.run_usage
* was non-NULL at write time) alongside the {jobid, state} header,
* so a promoted standby's catch-up replay recovers the
* client-reported per-host CPU/mem usage too, not just state. A
* payload of exactly sizeof(wal_jobstate_t) (no trailing bytes --
* any WAL_JOB_RUSAGE entry written before this fix, or one fired
* while job.run_usage was still NULL) decodes no run_usage,
* matching the pre-fix state-only behavior exactly. */
if (payload_len >= sizeof(wal_jobstate_t)) {
const wal_jobstate_t* js = (const wal_jobstate_t*)payload;
xce_job_runusage_t* runusage = NULL;
if (payload_len > sizeof(wal_jobstate_t)) {
runusage = decode_jobrunusage(
(const uint8_t*)payload + sizeof(wal_jobstate_t),
payload_len - sizeof(wal_jobstate_t));
}
restore_job_rusage_state(p->cluster, js->jobid, js->state,
runusage);
}
break;
}
case WAL_JOB_START: {
/* bug-1687: split out of the shared group above -- mirrors
* WAL_JOB_DISPATCH's split just below (bug-1678). This is the one
@ -4001,6 +4112,41 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn,
lock(j, RTTI_LOCK_WRITE);
job_transition_result_t result =
apply_job_update_transition(j, (int)js->state);
/* bug-1690: WAL_JOB_RUSAGE carries the client-reported run_usage
* data independent of the job's state transition -- a rusage
* report almost never changes state (the job stays RUN across
* every repeat report), so gating this adoption on
* result==JOB_TRANSITION_APPLIED the way START/FINISH/DISPATCH do
* below would make it dead code for that realistic, common case:
* every report after the first would see JOB_TRANSITION_NOOP_
* DUPLICATE and the decode would never run. Decode/adopt
* unconditionally, before the transition-guarded block below,
* whether or not the state actually changed. A payload of exactly
* sizeof(wal_jobstate_t) (no trailing bytes -- any WAL_JOB_RUSAGE
* entry written before this fix, or one fired while job.run_usage
* was still NULL at write time) decodes no run_usage, leaving it
* untouched, matching the pre-fix behavior exactly.
*
* MERGE, not free-and-replace: on the primary, this exact branch
* re-runs via persist_apply_pump_run()'s live pump moments after
* update_job_rusages() already live-mutated j->job.run_usage in
* place (with an xstring-backed host_name, makeString(),
* src/lib/xstring.c) -- j->job.run_usage here can genuinely be that
* SAME live structure, not just a replay artifact. Freeing it via
* the capnp-generated free_JobRunusage_ptr() (written for plain
* malloc'd, capnp-decoded Text fields) would free a non-malloc'd,
* RTTI-managed pointer -- heap corruption. merge_job_runusage()
* (job.c) never frees an existing host_name; see its own doc
* comment for the full rationale. */
if (wal_type == WAL_JOB_RUSAGE && len > sizeof(wal_jobstate_t)) {
xce_job_runusage_t* decoded = decode_jobrunusage(
(const uint8_t*)payload + sizeof(wal_jobstate_t),
len - sizeof(wal_jobstate_t));
if (decoded != NULL)
merge_job_runusage(&(j->job.run_usage), decoded);
}
if (result == JOB_TRANSITION_APPLIED) {
/* apply_job_update_transition() only carries {jobid, state} on
* this wire shape (see the doc comment above) and never touches
@ -7236,6 +7382,14 @@ int persist_job_hook_fun(int type, int situation, int detail,
/* All other job events: payload = wal_jobstate_t {jobid, state}.
* State is read from the job_t after the AFTER_EVENT hook updates it. */
wal_jobstate_t js = {0};
/* bug-1690: WAL_JOB_RUSAGE's own optional trailing Cap'n Proto
* JobRunusage bytes -- populated only by the HOOK_DETAIL_JOB_RUSAGE_UPDATE
* arm below, consumed by the shared append at the bottom of this
* function. Every other detail leaves this NULL, so their payload stays
* the plain wal_jobstate_t it always was -- zero behavior change for
* them. */
uint8_t* rusage_buf = NULL;
size_t rusage_len = 0;
switch (detail) {
case HOOK_DETAIL_JOB_START: {
/* args: cluster_t*, job_pidinfo_t*, job_t* */
@ -7286,6 +7440,22 @@ int persist_job_hook_fun(int type, int situation, int detail,
if (j == NULL)
return 0; /* job gone by the time this fired -- nothing to log */
js.jobid = rusages->jobid; js.state = (uint32_t)j->job.state;
/* bug-1690: WAL_JOB_RUSAGE previously carried only {jobid, state}
* -- the same silent-field-drop bug class as bug-1678/1687/1689 --
* the client-reported per-host CPU/mem usage (j->job.run_usage,
* just mutated in-place by update_job_rusages() moments ago, under
* the same write lock this AFTER_EVENT fires inside) was never in
* the log at all, so a restart replay left job.run_usage
* permanently NULL. Encode it as a trailing Cap'n Proto JobRunusage
* blob when present -- it is NULL only in test doubles that stub
* out update_host_runusage (e.g. test_commit_barrier_reply.c); the
* real production caller always has one by the time this fires. A
* NULL run_usage falls back to the legacy state-only payload at
* this function's shared tail, unchanged. */
if (j->job.run_usage != NULL)
(void)capnp_encode_jobrunusage(j->job.run_usage, &rusage_buf,
&rusage_len);
break;
}
default: {
@ -7298,6 +7468,25 @@ int persist_job_hook_fun(int type, int situation, int detail,
}
/* mode == WAL_MODE_RETAIN — fact-shaped job state transition */
if (rusage_buf != NULL) {
/* bug-1690: append the header + trailing JobRunusage bytes as one
* contiguous payload -- persist_submit_job_dispatch_fact()'s/
* persist_submit_job_start_fact()'s own header+trailer construction
* (bug-1678/1687), inlined here since WAL_JOB_RUSAGE is hook-driven
* (from the raw rusages payload update_job_rusages() fires, not a
* direct persist_submit_job_*_fact() call site job.c could own). */
size_t total = sizeof(js) + rusage_len;
uint8_t* payload = malloc(total);
int rc;
if (!payload) { free(rusage_buf); return -ENOMEM; }
memcpy(payload, &js, sizeof(js));
memcpy(payload + sizeof(js), rusage_buf, rusage_len);
free(rusage_buf);
rc = wal_hook_append_classified(p, wal_type, mode, payload,
(uint32_t)total);
free(payload);
return rc;
}
return wal_hook_append_classified(p, wal_type, mode, &js, sizeof(js));
}

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

@ -86,6 +86,7 @@ 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_wal_job_rusage_fact(void);
extern void test_job_dispatch_fail_count(void);
extern void test_apply_reapply_idempotency(void);
extern void test_apply_correctness_verification(void);
@ -258,6 +259,7 @@ int main(int argc, char* argv[])
TEST(test_wal_job_dispatch_fact);
TEST(test_wal_job_start_fact);
TEST(test_wal_job_finish_fact);
TEST(test_wal_job_rusage_fact);
TEST(test_job_dispatch_fail_count);
TEST(test_apply_reapply_idempotency);
TEST(test_apply_correctness_verification);

912
src/controller/testsuite/test_wal_job_rusage_fact.c Normal file
View file

@ -0,0 +1,912 @@
/* 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 <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_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);
}

31
src/include/wal_types.h
View file

@ -134,7 +134,36 @@ typedef enum {
WAL_JOB_STOP = 5,
WAL_JOB_RESUME = 6,
WAL_JOB_UPDATE = 7,
WAL_JOB_RUSAGE = 8,
WAL_JOB_RUSAGE = 8, /* client-reported per-host CPU/mem usage update.
* Payload is wal_jobstate_t{jobid, state, _pad=0}
* for any entry written before bug-1690, or one
* fired while job.run_usage was still NULL at
* write time (e.g. a test double stubbing out
* update_host_runusage); as of bug-1690
* persist_job_hook_fun's HOOK_DETAIL_JOB_RUSAGE_
* UPDATE case appends trailing Cap'n Proto
* JobRunusage bytes (the job's accumulated
* xce_job_runusage_t) whenever job.run_usage is
* non-NULL -- both apply seams treat a payload of
* exactly sizeof(wal_jobstate_t) as the legacy
* state-only shape for full backward
* compatibility. Unlike WAL_JOB_START/FINISH/
* DISPATCH, run_usage adoption on the
* apply_committed_entry() seam is NOT gated on a
* forward state transition -- a rusage report
* almost never changes job.state, so gating it
* the same way would make it dead code for the
* realistic, repeated-report case. Also unlike
* those three, adoption is a MERGE
* (merge_job_runusage(), job.c), never a
* free-and-replace: job.run_usage can be the
* job's LIVE, in-place-mutated structure (an
* xstring-backed host_name from makeString(),
* not a plain malloc()) when the primary's own
* live apply pump re-applies a fact it just
* wrote -- freeing it via the capnp-generated
* free_JobRunusage_ptr() would corrupt the
* heap. */
WAL_JOB_CKPT = 9,
WAL_JOB_PREASON = 10,
WAL_JOB_ARRAY_SUBMIT = 11, /* job-array-single-wal-record: one record for

29
src/schema/controller.capnp
View file

@ -266,6 +266,35 @@ struct JobPIDInfo $C.mapname("job_pidinfo_t"){
exdata @3: Text $C.mapname("exdata");
}
# bug-1690: xce_job_runusage_t (xce/types.h) previously had NO Cap'n Proto
# mapping at all -- unlike JobAlloc/JobPIDInfo above, which already existed
# as Job sub-fields the WAL_JOB_DISPATCH/WAL_JOB_START fixes (bug-1678/1687)
# could reuse standalone. WAL_JOB_RUSAGE needs one to carry the client-
# reported per-host CPU/mem usage in its fact payload so a restart/catch-up
# replay can reconstruct job.run_usage instead of leaving it permanently
# NULL. NOT the same type as JobRusages/SchedJobRusages below (xce_job_
# rusages_t) -- that is the client's WIRE UPDATE REQUEST shape
# ({jobid, pids[], pgids[], usages[]}); this is the CONTROLLER'S ACCUMULATED
# per-job/per-host usage shape, a different struct entirely.
struct HostRunusage $C.mapname("xce_host_runusage_t"){
hostName @0: Text $C.mapname("host_name");
cpuTime @1: Int32 $C.mapname("cpu_time");
memUsage @2: Float64 $C.mapname("mem_usage");
}
struct JobRunusage $C.mapname("xce_job_runusage_t"){
maxMem @0: Float64 $C.mapname("max_mem");
avgMem @1: Float64 $C.mapname("avg_mem");
usedMem @2: Float64 $C.mapname("used_mem");
cpuTime @3: Int32 $C.mapname("cpu_time");
peakCpuTime @4: Int32 $C.mapname("peak_cpu_time");
peakCpuDuration @5: Int32 $C.mapname("peak_cpu_duration");
memEfficiency @6: Float32 $C.mapname("mem_efficiency");
avgCpuEfficiency @7: Float32 $C.mapname("avg_cpu_efficiency");
peakCpuEfficiency @8: Float32 $C.mapname("peak_cpu_efficiency");
hostRunusage @9: List(HostRunusage) $C.mapname("host_runusage") $C.maplistcount("num_hosts");
}
struct JobFail $C.mapname("job_failinfo_t"){
jobId @0: UInt64 $C.mapname("jobid");
code @1: Int32 $C.mapname("code");