From f7b732a30070efdb779f2f30b1ae85932a356548 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 25 Aug 2026 10:53:32 +0000 Subject: [PATCH] fix(controller): WAL_JOB_RUSAGE loses job.run_usage on replay (bug-1690) 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 Claude-Session: https://claude.ai/code/session_01AgYRU3zDTqNGV5Jk8aiKLg --- .wolf/buglog.json | 31 + src/controller/job.c | 133 +++ src/controller/meson.build | 1 + src/controller/persist.c | 191 +++- src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_wal_job_rusage_fact.c | 912 ++++++++++++++++++ src/include/wal_types.h | 31 +- src/schema/controller.capnp | 29 + 8 files changed, 1328 insertions(+), 2 deletions(-) create mode 100644 src/controller/testsuite/test_wal_job_rusage_fact.c diff --git a/.wolf/buglog.json b/.wolf/buglog.json index cd776633..dcfae1a5 100644 --- a/.wolf/buglog.json +++ b/.wolf/buglog.json @@ -18843,6 +18843,37 @@ ], "occurrences": 1, "last_seen": "2026-08-25T18:10:00Z" + }, + { + "id": "bug-1690", + "timestamp": "2026-08-25T11:00:00Z", + "error_message": "xce-audit-invariants finding: WAL_JOB_RUSAGE silently drops job.run_usage (client-reported per-host CPU/mem usage) on a controller restart replay -- same bug class as bug-1678 (WAL_JOB_DISPATCH/job.allocs), bug-1687 (WAL_JOB_START/job.pidinfo+start_time), bug-1689 (WAL_JOB_FINISH/job.exit_info+finish_time). update_job_rusages() (job.c) sets job->job.run_usage (xce_job_runusage_t*, an accumulated per-host struct) via update_host_runusage() (misc.c), but the WAL_JOB_RUSAGE fact persist_job_hook_fun (persist.c) wrote only carried wal_jobstate_t{jobid, state} -- run_usage was never in the log at all, so a restart replay left it permanently NULL even though a client actively reported CPU/mem usage. Unlike the three precedents, xce_job_runusage_t had NO existing Cap'n Proto mapping at all (confirmed: grep for xce_job_runusage_t in controller.capnp matched nothing; a similar-looking but unrelated type, xce_job_rusages_t/JobRusages/SchedJobRusages -- the CLIENT's wire update-request shape {jobid, pids[], pgids[], usages[]} -- already existed and is NOT interchangeable), so this fix required a genuine schema addition, not a reuse.", + "file": "src/schema/controller.capnp (new HostRunusage/JobRunusage structs), src/controller/persist.c (capnp_encode_jobrunusage/decode_jobrunusage ~L327-388, persist_job_hook_fun's HOOK_DETAIL_JOB_RUSAGE_UPDATE case ~L7440-7500, apply_committed_entry's WAL_JOB_RUSAGE branch ~L4116-4144, apply_wal_entry's WAL_JOB_RUSAGE case ~L3712-3740), src/controller/job.c (merge_job_runusage new ~L893-1000, restore_job_rusage_state new), src/include/wal_types.h (WAL_JOB_RUSAGE doc comment), src/controller/testsuite/test_wal_job_rusage_fact.c (new, 8 tests)", + "root_cause": "persist_job_hook_fun()'s HOOK_DETAIL_JOB_RUSAGE_UPDATE case (persist.c) built only wal_jobstate_t{jobid, state} for the WAL_JOB_RUSAGE fact, discarding the job's run_usage entirely even though it was available (looked up via the same jobmgr lookup, under the same write lock update_job_rusages() already held). Both apply seams -- apply_committed_entry()'s shared WAL_JOB_START/FINISH/.../DISPATCH fall-through group (persist.c) and apply_wal_entry()'s identical shared group -> restore_job_state() (job.c, state-only) -- never restored run_usage, matching the exact same class of gap bug-1678/1687/1689 already fixed for allocs/pidinfo+start_time/exit_info+finish_time. Confirmed via CodeGraph source read of update_job_rusages() (job.c) and persist_job_hook_fun()'s actual HOOK_DETAIL_JOB_RUSAGE_UPDATE arm (not a direct persist_submit_job_*_fact() call site like DISPATCH/START/FINISH have -- RUSAGE is purely hook-driven from the raw xce_job_rusages_t* payload update_job_rusages() fires, a genuine divergence from the audit's initial reconstructed precedent that was verified against the real code before implementing, per this task's explicit instruction).", + "fix": "FIXED (2026-08-25). Added a genuine new Cap'n Proto schema pair to controller.capnp -- struct HostRunusage $C.mapname(\"xce_host_runusage_t\"){hostName, cpuTime, memUsage} and struct JobRunusage $C.mapname(\"xce_job_runusage_t\"){maxMem, avgMem, usedMem, cpuTime, peakCpuTime, peakCpuDuration, memEfficiency, avgCpuEfficiency, peakCpuEfficiency, hostRunusage: List(HostRunusage) $C.maplistcount(\"num_hosts\")} -- following JobPendReason's exact List(NestedStruct)+maplistcount precedent already in the schema, field types verified against types.h's real xce_job_runusage_t/xce_host_runusage_t definitions via CodeGraph before writing. The meson custom_target (src/lib/meson.build) regenerated lib/controller.capnp.c/.h automatically on rebuild -- verified via nm on the built controller_unit_test binary showing the full genuinely-codegen'd symbol set (new_JobRunusage, read/write_JobRunusage, encode/decode_JobRunusage, free_JobRunusage_ptr, get/set per-field accessors, and the equivalent HostRunusage set), never hand-written. persist.c gains capnp_encode_jobrunusage()/decode_jobrunusage() (mirroring capnp_encode_joballoc()/decode_joballoc() exactly). persist_job_hook_fun()'s HOOK_DETAIL_JOB_RUSAGE_UPDATE case now also encodes j->job.run_usage (when non-NULL) as trailing capnp bytes after the wal_jobstate_t header -- inlined into that hook (not a new persist_submit_job_rusage_fact(), unlike bug-1678/1687/1689's direct-call-site pattern, since RUSAGE is genuinely hook-driven). Both apply seams updated: apply_wal_entry() splits WAL_JOB_RUSAGE out of its shared state-only group into its own case, decoding trailing bytes and calling new restore_job_rusage_state() (job.c). apply_committed_entry()'s WAL_JOB_RUSAGE handling stays in its shared job-fact group but -- critically, UNLIKE START/FINISH/DISPATCH -- decodes/adopts run_usage UNCONDITIONALLY, before the JOB_TRANSITION_APPLIED gate: a rusage report almost never changes job.state (the job stays RUN across every repeat report), so gating adoption on a forward state transition the way the other three fixes do would make it dead code for the realistic, common case (every report after the first hits JOB_TRANSITION_NOOP_DUPLICATE and the gated block never runs). This divergence was caught by actually reading apply_job_update_transition()'s guard logic rather than blindly copying the precedent pattern. A payload of exactly sizeof(wal_jobstate_t) (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 on both seams, matching pre-fix behavior exactly -- full backward compatibility.\n\nCRITICAL SAFETY FINDING beyond the precedent's pattern, caught via investigation (not assumed): a naive free-and-replace (mirroring bug-1678/1687/1689's free_X_ptr()-then-adopt exactly) would be a genuine heap-corruption hazard here, specific to RUSAGE. 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 BEFORE the WAL fact is even appended, and host_name is sourced from makeString() (src/lib/xstring.c), an RTTI-managed, refcounted, header-before-data pointer (confirmed via librtti's actual source in the nix store: RTTI_DATA_TO_OBJ subtracts RTTI_OBJ_HEADER_SIZE from the data pointer to reach the real allocation start) -- NOT a plain malloc() allocation, unlike JobAlloc/JobPIDInfo's host/exdata fields (verified to be genuinely capnp-wire-decoded end to end via dispatch_a_job_()'s own alloc parameter, confirming why bug-1678/1687 never hit this). persist_apply_pump_run() (the live online-apply pump, confirmed via its own doc comment to run on the primary itself, not just boot replay) re-applies every committed WAL_JOB_RUSAGE fact via apply_committed_entry() moments after the SAME live mutation already happened -- so calling the capnp-generated free_JobRunusage_ptr() (written for plain malloc'd Text fields) on that live structure would free() a non-malloc'd, RTTI-managed pointer: guaranteed ASAN abort / heap corruption on the very first live rusage report for a new host. Fixed by making adoption a MERGE (new merge_job_runusage(), job.c) instead of a free-and-replace: an already-present host (matched by host_name string comparison, mirroring update_host_runusage()'s own find-existing-or-append semantics) only has its scalar cpu_time/mem_usage overwritten in place -- its host_name pointer, possibly xstring-backed, is NEVER touched or freed; only a genuinely new host's own freshly-decoded (always plain-malloc'd) host_name is adopted by ownership transfer. Applied identically in both restore_job_rusage_state() (apply_wal_entry seam) and apply_committed_entry's own branch, since a demoted-primary-turned-standby could in principle retain live-mutated run_usage data too. Noted (not fixed, genuinely out of scope -- a different file/component/bug class): job_destroy() (src/lib/jobj.c)'s own existing manual FREEUP-based walk over run_usage has this exact same latent host_name/xstring mismatch, independent of and pre-dating this fix.\n\nRegression tests: src/controller/testsuite/test_wal_job_rusage_fact.c (8 tests, mirroring test_wal_job_start_fact.c's structure): (1) apply_committed_entry adopts run_usage when state does NOT change (JOB_TRANSITION_NOOP_DUPLICATE) -- the suite's most important test, proving the unconditional-decode design choice; (2) apply_committed_entry adopts run_usage on a genuine forward transition, hook still fires once; (3) apply_committed_entry backward-compat with an old-format payload; (4) apply_committed_entry merges without freeing a REAL makeString()'d host_name -- builds a live-flavor run_usage by hand, applies an update for that same host plus a new host, and asserts the host_name POINTER IDENTITY survives (proving it was never freed) alongside correct scalar/new-host merging, with careful test-only cleanup (detach + dereference()) so the separate, pre-existing job_destroy() FREEUP issue can't fire inside the test's own teardown; (5) apply_wal_entry adopts run_usage, no hook, absolute state overwrite; (6) apply_wal_entry backward-compat; (7) direct-payload restart-replay (WAL_JOB_SUBMIT + WAL_JOB_RUSAGE via persist_wal_async + persist_recover()); (8) drives the REAL, unmodified persist_job_hook_fun() via the established 'variadic trampoline' pattern (test_commit_barrier_rollout.c/test_persist_hooks.c) through a real WAL-write + persist_recover() restart cycle -- NOT via update_job_rusages() itself, because this same controller_unit_test binary already links test_commit_barrier_reply.c's own non-static stub of update_host_runusage() (a no-op, since misc.c is outside controller_unit_test_src), so calling update_job_rusages() here would silently hit that stub and populate nothing -- documented in the test file's own header comment.\n\nRED->GREEN independently verified live (not narrated): `git stash push -- src/controller/job.c src/controller/persist.c` (keeping the schema change, the new test file, and meson.build/controller_test.c wiring), full clean rebuild via `nix develop ./src --command bash src/tools/xce-build.sh controller asan build-asan` (no --incremental, since this is the schema's first-ever compile) -- 1160 Tests 8 Failures: exactly the 6 new RUSAGE tests (5 with the exact expected 'must set job.run_usage' mismatch, 1 with 'Expected 500 Was 100' for the merge-safety test) plus the 2 pre-existing bug-1561 test_config_repl.c failures; the 2 backward-compat tests correctly passed unchanged. `git stash pop`, incremental rebuild -- 1160 Tests 2 Failures (only the pre-existing bug-1561 pair, exact signature match 'Expected 2 Was 3' / 'Expected 0 Was '), all 8 new tests pass, zero new failures, zero ASAN aborts/heap-corruption findings (explicitly grepped the full test log for AddressSanitizer/heap-buffer-overflow/invalid-free/SEGV/Aborted -- none found). Verified the generated capnp code was genuinely regenerated, not hand-written: `nm` on the built controller_unit_test binary shows the complete real codegen symbol set for both new structs (new_JobRunusage, read/write_JobRunusage, encode/decode_JobRunusage, free_JobRunusage_ptr, per-field get/set accessors, and the equivalent HostRunusage set), with generated-file mtimes strictly after the schema edit's mtime.", + "tags": [ + "controller", + "job.c", + "persist.c", + "wal", + "wal-job-rusage", + "job-run-usage", + "restart-recovery", + "data-loss", + "apply-seam", + "capnp-schema-change", + "heap-corruption-hazard", + "xstring-rtti-pointer", + "merge-not-replace" + ], + "related_bugs": [ + "bug-1678", + "bug-1687", + "bug-1689", + "bug-1561" + ], + "occurrences": 1, + "last_seen": "2026-08-25T11:00:00Z" } ] } diff --git a/src/controller/job.c b/src/controller/job.c index ea9f5317..3778e456 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -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; diff --git a/src/controller/meson.build b/src/controller/meson.build index 5c60c79a..f140c251 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -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', diff --git a/src/controller/persist.c b/src/controller/persist.c index 569be89c..5dd90a83 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -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)); } diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 26c8cf28..e4fd4ebb 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -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); diff --git a/src/controller/testsuite/test_wal_job_rusage_fact.c b/src/controller/testsuite/test_wal_job_rusage_fact.c new file mode 100644 index 00000000..dc83187c --- /dev/null +++ b/src/controller/testsuite/test_wal_job_rusage_fact.c @@ -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 + +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "cluster.h" +#include "controller.capnp.h" +#include "hook.h" +#include "jobj.h" +#include "persist.h" +#include "wal_types.h" +#include "xce/cluster_def.h" +#include "xce/types.h" +#include "xcealloc.h" +#include "xstring.h" + +/* jobj.c's jobsub table and the process-global xstring table are shared + * across every test file in the controller_unit_test binary — guard + * idempotently, mirroring every sibling fixture in this suite. */ +static int g_wru_jobobj_ready; + +static cluster_t g_wru_c; +static persist_t g_wru_p; + +/* ── Spy hook: counts AFTER_EVENT firings, ignores all payload args ─────── */ +static int g_wru_hook_fires; + +static int wru_spy_hook(int type, int situation, int detail, void* data, + va_list ap) +{ + (void)type; (void)detail; (void)data; (void)ap; + if (situation == HOOK_AFTER_EVENT) + g_wru_hook_fires++; + return 0; +} + +static void wru_setup(void) +{ + kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, + .nslots = 32, + .delta = 16 }; + + if (makeString((char*)"wru-probe") == NULL) + xstring_table_init(64); + + if (!g_wru_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_wru_jobobj_ready = 1; + } + + memset(&g_wru_c, 0, sizeof(g_wru_c)); + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&g_wru_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); + + g_wru_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + TEST_ASSERT_NOT_NULL(g_wru_c.jobmgr); + + memset(&g_wru_p, 0, sizeof(g_wru_p)); + g_wru_p.cluster = &g_wru_c; + + g_wru_hook_fires = 0; +} + +static void wru_teardown(void) +{ + xce_hook_fini(&g_wru_c.hooksys); + if (g_wru_c.jobmgr) + dereference(g_wru_c.jobmgr); +} + +/* Look up a job_t wrapper by jobid (NULL if absent). */ +static job_t* wru_find_job(uint64_t jobid) +{ + job_t* j = NULL; + lock(g_wru_c.jobmgr, RTTI_LOCK_READ); + int found = key_find(g_wru_c.jobmgr, (void*)jobid, (void**)&j); + unlock(g_wru_c.jobmgr); + return (found > 0) ? j : NULL; +} + +/* Creates a job directly in the fixture's jobmgr, bypassing dispatch/replay + * entirely — for seeding fact-transition tests, mirroring + * test_wal_job_dispatch_fact.c's wjd_seed_job / test_wal_job_start_fact.c's + * wjs_seed_job. */ +static job_submit_data_t g_wru_seed_submit; + +static job_t* wru_seed_job(uint64_t jobid, uint32_t state) +{ + xce_job_t jdef; + + memset(&g_wru_seed_submit, 0, sizeof(g_wru_seed_submit)); + g_wru_seed_submit.queue = (char*)"q"; + g_wru_seed_submit.user = (char*)"tester"; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = state; + jdef.submit = &g_wru_seed_submit; + + return makeJob(g_wru_c.jobmgr, &jdef); +} + +/* ── Real fixture data: 2 hosts' worth of accumulated CPU/mem usage ──────── */ + +static xce_host_runusage_t g_wru_host_a = { + .host_name = (char*)"wru-host-a", .cpu_time = 1200, .mem_usage = 512.5 +}; +static xce_host_runusage_t g_wru_host_b = { + .host_name = (char*)"wru-host-b", .cpu_time = 900, .mem_usage = 256.25 +}; +static xce_host_runusage_t* g_wru_host_ptrs[2]; + +static xce_job_runusage_t wru_make_test_runusage(void) +{ + xce_job_runusage_t r; + + memset(&r, 0, sizeof(r)); + r.max_mem = 1024.5; + r.avg_mem = 768.25; + r.used_mem = 900.0; + r.cpu_time = 2100; + r.peak_cpu_time = 1500; + r.peak_cpu_duration = 30; + r.mem_efficiency = 0.75f; + r.avg_cpu_efficiency = 0.6f; + r.peak_cpu_efficiency = 0.9f; + r.num_hosts = 2; + + g_wru_host_ptrs[0] = &g_wru_host_a; + g_wru_host_ptrs[1] = &g_wru_host_b; + r.host_runusage = g_wru_host_ptrs; + + return r; +} + +static void wru_assert_runusage_matches(xce_job_runusage_t* got) +{ + TEST_ASSERT_NOT_NULL_MESSAGE(got, + "a WAL_JOB_RUSAGE payload with trailing JobRunusage bytes must set " + "job.run_usage (bug-1690), not just state"); + TEST_ASSERT_EQUAL_FLOAT(1024.5, got->max_mem); + TEST_ASSERT_EQUAL_FLOAT(768.25, got->avg_mem); + TEST_ASSERT_EQUAL_FLOAT(900.0, got->used_mem); + TEST_ASSERT_EQUAL_INT(2100, got->cpu_time); + TEST_ASSERT_EQUAL_INT(1500, got->peak_cpu_time); + TEST_ASSERT_EQUAL_INT(30, got->peak_cpu_duration); + TEST_ASSERT_EQUAL_FLOAT(0.75f, got->mem_efficiency); + TEST_ASSERT_EQUAL_FLOAT(0.6f, got->avg_cpu_efficiency); + TEST_ASSERT_EQUAL_FLOAT(0.9f, got->peak_cpu_efficiency); + TEST_ASSERT_EQUAL_INT(2, got->num_hosts); + TEST_ASSERT_NOT_NULL(got->host_runusage); + TEST_ASSERT_EQUAL_STRING("wru-host-a", got->host_runusage[0]->host_name); + TEST_ASSERT_EQUAL_INT(1200, got->host_runusage[0]->cpu_time); + TEST_ASSERT_EQUAL_FLOAT(512.5, got->host_runusage[0]->mem_usage); + TEST_ASSERT_EQUAL_STRING("wru-host-b", got->host_runusage[1]->host_name); + TEST_ASSERT_EQUAL_INT(900, got->host_runusage[1]->cpu_time); + TEST_ASSERT_EQUAL_FLOAT(256.25, got->host_runusage[1]->mem_usage); +} + +/* Build a wal_jobstate_t header + trailing Cap'n Proto JobRunusage bytes -- + * persist_job_hook_fun's WAL_JOB_RUSAGE payload shape -- mirroring + * wjs_build_start_payload() (test_wal_job_start_fact.c) since persist.c's + * own static capnp_encode_jobrunusage() isn't linkable from a test TU. + * Caller owns *out. */ +static int wru_build_rusage_payload(uint64_t jobid, uint32_t state, + xce_job_runusage_t* r, + uint8_t** out, size_t* out_len) +{ + struct capn c; + struct capn_segment* cs; + struct JobRunusage cjr; + JobRunusage_ptr jrp; + uint8_t* rbuf; + size_t rlen; + wal_jobstate_t js = { .jobid = jobid, .state = state, ._pad = 0 }; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_JobRunusage(cs, &cjr, r); + jrp = new_JobRunusage(cs); + write_JobRunusage(&cjr, jrp); + capn_setp(capn_root(&c), 0, jrp.p); + + rlen = capn_size(&c); + rbuf = malloc(rlen); + if (!rbuf) { capn_free(&c); return -ENOMEM; } + capn_write_mem(&c, rbuf, rlen, 0); + capn_free(&c); + + *out_len = sizeof(js) + rlen; + *out = malloc(*out_len); + if (!*out) { free(rbuf); return -ENOMEM; } + memcpy(*out, &js, sizeof(js)); + memcpy(*out + sizeof(js), rbuf, rlen); + free(rbuf); + return 0; +} + +/* ── apply_committed_entry(): live-commit apply / boot replay dispatch ──── */ + +/* THE key regression test: seeds the job already RUNNING and replays a + * WAL_JOB_RUSAGE fact whose own state is ALSO RUN -- the realistic, + * overwhelmingly common shape of a real rusage report (job.state never + * actually changes across repeat reports). apply_job_update_transition() + * reports JOB_TRANSITION_NOOP_DUPLICATE for this (state == cur), so a fix + * that (incorrectly) gated run_usage adoption on JOB_TRANSITION_APPLIED -- + * mirroring WAL_JOB_START/FINISH/DISPATCH's own gated fields -- would leave + * job.run_usage NULL here. This test fails against that naive shape and + * passes against the actual fix (unconditional decode/adopt). */ +static void test_wru_apply_committed_entry_sets_runusage_no_state_change(void) +{ + wru_setup(); + + job_t* j = wru_seed_job(73001, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j); + + xce_job_runusage_t r = wru_make_test_runusage(); + + uint8_t* payload = NULL; + size_t plen = 0; + TEST_ASSERT_EQUAL_INT(0, + wru_build_rusage_payload(73001, JOB_STAT_RUN, &r, &payload, &plen)); + + xce_add_hook(&g_wru_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wru_spy_hook, + NULL); + + int rc = apply_committed_entry(&g_wru_p, WAL_JOB_RUSAGE, 300, payload, + (uint32_t)plen); + free(payload); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); + wru_assert_runusage_matches(j->job.run_usage); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wru_hook_fires, + "a same-state (duplicate) RUSAGE replay must NOT fire a " + "notification hook -- unchanged from before this fix -- even though " + "run_usage WAS adopted"); + + wru_teardown(); + printf("test_wru_apply_committed_entry_sets_runusage_no_state_change: " + "PASS\n"); +} + +/* Secondary scenario: a WAL_JOB_RUSAGE fact that DOES carry a forward state + * transition (mechanism-level test, mirroring how the START/DISPATCH suites + * use PEND->RUN generically). run_usage must be adopted here too, and the + * existing forward-transition hook-fire behavior must be unchanged. */ +static void +test_wru_apply_committed_entry_sets_runusage_forward_transition(void) +{ + wru_setup(); + + job_t* j = wru_seed_job(73002, JOB_STAT_PEND); + TEST_ASSERT_NOT_NULL(j); + + xce_job_runusage_t r = wru_make_test_runusage(); + + uint8_t* payload = NULL; + size_t plen = 0; + TEST_ASSERT_EQUAL_INT(0, + wru_build_rusage_payload(73002, JOB_STAT_RUN, &r, &payload, &plen)); + + xce_add_hook(&g_wru_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wru_spy_hook, + NULL); + + int rc = apply_committed_entry(&g_wru_p, WAL_JOB_RUSAGE, 301, payload, + (uint32_t)plen); + free(payload); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); + wru_assert_runusage_matches(j->job.run_usage); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_wru_hook_fires, + "a forward RUSAGE-carried transition must still fire exactly one " + "notification, unchanged from before this fix"); + + wru_teardown(); + printf( + "test_wru_apply_committed_entry_sets_runusage_forward_transition: " + "PASS\n"); +} + +static void +test_wru_apply_committed_entry_backward_compat_old_payload(void) +{ + wru_setup(); + + job_t* j = wru_seed_job(73003, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j); + TEST_ASSERT_NULL_MESSAGE(j->job.run_usage, + "fixture must seed a job with run_usage still unset"); + + wal_jobstate_t js = { .jobid = 73003, .state = JOB_STAT_RUN, ._pad = 0 }; + + int rc = apply_committed_entry(&g_wru_p, WAL_JOB_RUSAGE, 302, &js, + sizeof(js)); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); + TEST_ASSERT_NULL_MESSAGE(j->job.run_usage, + "an old-format (exactly sizeof(wal_jobstate_t)) WAL_JOB_RUSAGE " + "payload must leave run_usage NULL -- no crash, no misread of " + "absent trailing bytes"); + + wru_teardown(); + printf("test_wru_apply_committed_entry_backward_compat_old_payload: " + "PASS\n"); +} + +/* ── Merge safety: never free a LIVE (possibly xstring-backed) host_name ── + * + * bug-1690's most important safety finding: update_job_rusages()/ + * update_host_runusage() (job.c/misc.c) build the LIVE job.run_usage with + * host_name sourced from makeString() (src/lib/xstring.c) -- an + * RTTI-managed, header-before-data pointer (RTTI_DATA_TO_OBJ, librtti), NOT + * a plain malloc() allocation. apply_committed_entry()'s WAL_JOB_RUSAGE + * branch can run against this EXACT live structure -- the primary's own + * live apply pump (persist_apply_pump_run()) re-applies a fact moments + * after the fact's own live mutation already happened. A naive + * free-and-replace (the capnp-generated free_JobRunusage_ptr(), written for + * plain malloc'd Text fields) would free() a non-malloc'd pointer there: + * guaranteed heap corruption under ASAN. This test builds a run_usage with + * a REAL makeString()'d host_name (mirroring the live-mutation flavor + * exactly), then applies a WAL_JOB_RUSAGE fact reporting an update for that + * SAME host plus a brand-new second host, and asserts: (a) no crash + * (implicit -- ASAN aborts mid-test on a corrupting free), (b) the existing + * host's host_name POINTER IDENTITY survives (proving it was merged in + * place, never freed+reallocated), (c) its cpu_time/mem_usage reflect the + * new report, (d) the new host is correctly appended. */ +static void +test_wru_apply_committed_entry_merges_without_freeing_live_host(void) +{ + wru_setup(); + + job_t* j = wru_seed_job(73008, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j); + + /* Build a LIVE-flavor run_usage by hand, mirroring update_host_ + * runusage()'s own construction exactly: calloc'd struct/array/entries, + * but a REAL makeString()'d host_name -- the exact allocator mismatch + * this test exists to catch. */ + char* live_host_name = makeString((char*)"wru-live-host"); + TEST_ASSERT_NOT_NULL(live_host_name); + + xce_job_runusage_t* live = (xce_job_runusage_t*)calloc(1, sizeof(*live)); + TEST_ASSERT_NOT_NULL(live); + live->max_mem = 100.0; + live->cpu_time = 50; + live->num_hosts = 1; + live->host_runusage = + (xce_host_runusage_t**)calloc(1, sizeof(xce_host_runusage_t*)); + TEST_ASSERT_NOT_NULL(live->host_runusage); + live->host_runusage[0] = + (xce_host_runusage_t*)calloc(1, sizeof(xce_host_runusage_t)); + TEST_ASSERT_NOT_NULL(live->host_runusage[0]); + live->host_runusage[0]->host_name = live_host_name; + live->host_runusage[0]->cpu_time = 111; + live->host_runusage[0]->mem_usage = 222.0; + + j->job.run_usage = live; + + /* Fact payload: an update for "wru-live-host" plus a brand-new + * "wru-new-host" -- exactly what a real subsequent rusage report for + * this job would carry. */ + xce_host_runusage_t update_host = { + .host_name = (char*)"wru-live-host", .cpu_time = 999, + .mem_usage = 888.0 + }; + xce_host_runusage_t new_host = { + .host_name = (char*)"wru-new-host", .cpu_time = 42, .mem_usage = 84.0 + }; + xce_host_runusage_t* host_ptrs[2] = { &update_host, &new_host }; + + xce_job_runusage_t r; + memset(&r, 0, sizeof(r)); + r.max_mem = 500.0; + r.cpu_time = 250; + r.num_hosts = 2; + r.host_runusage = host_ptrs; + + uint8_t* payload = NULL; + size_t plen = 0; + TEST_ASSERT_EQUAL_INT(0, + wru_build_rusage_payload(73008, JOB_STAT_RUN, &r, &payload, &plen)); + + int rc = apply_committed_entry(&g_wru_p, WAL_JOB_RUSAGE, 305, payload, + (uint32_t)plen); + free(payload); + + TEST_ASSERT_EQUAL_INT(0, rc); + + xce_job_runusage_t* got = j->job.run_usage; + TEST_ASSERT_NOT_NULL(got); + TEST_ASSERT_EQUAL_FLOAT(500.0, got->max_mem); + TEST_ASSERT_EQUAL_INT(250, got->cpu_time); + TEST_ASSERT_EQUAL_INT(2, got->num_hosts); + + int live_idx = -1, found_new = 0; + for (int i = 0; i < got->num_hosts; i++) { + xce_host_runusage_t* h = got->host_runusage[i]; + TEST_ASSERT_NOT_NULL(h); + if (h->host_name != NULL && + strcmp(h->host_name, "wru-live-host") == 0) { + live_idx = i; + TEST_ASSERT_TRUE_MESSAGE(h->host_name == live_host_name, + "an existing host's host_name pointer must be preserved by " + "identity -- merge must never free+replace it (bug-1690 " + "heap-corruption hazard: it may be an xstring, not a " + "malloc'd pointer)"); + TEST_ASSERT_EQUAL_INT(999, h->cpu_time); + TEST_ASSERT_EQUAL_FLOAT(888.0, h->mem_usage); + } else if (h->host_name != NULL && + strcmp(h->host_name, "wru-new-host") == 0) { + found_new = 1; + TEST_ASSERT_EQUAL_INT(42, h->cpu_time); + TEST_ASSERT_EQUAL_FLOAT(84.0, h->mem_usage); + } + } + TEST_ASSERT_TRUE_MESSAGE(live_idx >= 0, + "existing host must survive the merge"); + TEST_ASSERT_TRUE_MESSAGE(found_new, + "new host must be appended by the merge"); + + /* Cleanup: job_destroy() (jobj.c) releases run_usage via a manual + * FREEUP-based walk that (independent of this fix -- a separate, + * pre-existing, out-of-scope issue) assumes every host_name is plain + * malloc'd. Detach the still-live, still-xstring-backed host_name + * before teardown so THAT unrelated issue can't fire inside this + * test's own cleanup, and release it correctly ourselves. */ + if (live_idx >= 0) + got->host_runusage[live_idx]->host_name = NULL; + dereference(live_host_name); + + wru_teardown(); + printf( + "test_wru_apply_committed_entry_merges_without_freeing_live_host: " + "PASS\n"); +} + +/* ── apply_wal_entry(): standby catch-up / replicated-apply dispatch ────── */ + +static void test_wru_apply_wal_entry_sets_runusage_no_hook(void) +{ + wru_setup(); + + job_t* j = wru_seed_job(73004, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j); + + xce_job_runusage_t r = wru_make_test_runusage(); + + uint8_t* payload = NULL; + size_t plen = 0; + TEST_ASSERT_EQUAL_INT(0, + wru_build_rusage_payload(73004, JOB_STAT_RUN, &r, &payload, &plen)); + + xce_add_hook(&g_wru_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wru_spy_hook, + NULL); + + int rc = apply_wal_entry(&g_wru_p, WAL_JOB_RUSAGE, 303, payload, + (uint32_t)plen); + free(payload); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); + wru_assert_runusage_matches(j->job.run_usage); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wru_hook_fires, + "apply_wal_entry's catch-up dispatch must never fire a " + "notification hook"); + + wru_teardown(); + printf("test_wru_apply_wal_entry_sets_runusage_no_hook: PASS\n"); +} + +static void test_wru_apply_wal_entry_backward_compat_old_payload(void) +{ + wru_setup(); + + job_t* j = wru_seed_job(73005, JOB_STAT_RUN); + TEST_ASSERT_NOT_NULL(j); + TEST_ASSERT_NULL_MESSAGE(j->job.run_usage, + "fixture must seed a job with run_usage still unset"); + + wal_jobstate_t js = { .jobid = 73005, .state = JOB_STAT_RUN, ._pad = 0 }; + + int rc = apply_wal_entry(&g_wru_p, WAL_JOB_RUSAGE, 304, &js, sizeof(js)); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); + TEST_ASSERT_NULL_MESSAGE(j->job.run_usage, + "an old-format WAL_JOB_RUSAGE payload must leave run_usage NULL on " + "this seam, unchanged from before this fix"); + + wru_teardown(); + printf("test_wru_apply_wal_entry_backward_compat_old_payload: PASS\n"); +} + +/* ── Restart-replay Observables ──────────────────────────────────────── */ + +static persist_t* wru_make_fresh_persist(const char* tmpdir) +{ + persist_t* p = calloc(1, sizeof(persist_t)); + if (!p) return NULL; + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; /* slow fsync — no races in tests */ + + persist_set_workdir(p, tmpdir, "ctrl-test"); + + if (persist_init(p, &cfg, NULL) != 0) { + free(p); + return NULL; + } + if (persist_start(p) != 0) { + persist_fini(p); + free(p); + return NULL; + } + return p; +} + +static void wru_teardown_persist(persist_t* p) +{ + persist_stop(p); + persist_fini(p); + free(p); +} + +/* Encode helper mirroring test_wal_job_start_fact.c's wjs_encode_job + * (capnp_encode_job is file-static in persist.c). */ +static int wru_encode_job(const xce_job_t* j, uint8_t** out, size_t* out_len) +{ + struct capn c; + struct capn_segment* cs; + struct Job cj; + Job_ptr jp; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_Job(cs, &cj, (xce_job_t*)j); + jp = new_Job(cs); + write_Job(&cj, jp); + capn_setp(capn_root(&c), 0, jp.p); + + *out_len = capn_size(&c); + *out = malloc(*out_len); + if (!*out) { capn_free(&c); return -ENOMEM; } + capn_write_mem(&c, *out, *out_len, 0); + capn_free(&c); + return 0; +} + +/* Direct-payload restart replay: appends WAL_JOB_SUBMIT + a new-format + * WAL_JOB_RUSAGE (real run_usage, same-state RUN->RUN -- the realistic + * shape) straight onto disk via persist_wal_async, then replays via a fresh + * persist_t's persist_recover() (a real restart). Isolates the WAL fact + * shape from the live caller, mirroring test_wal_job_start_fact.c's own + * restart-replay test. */ +static void test_wru_restart_replay_recovers_runusage(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_wru_replay_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + /* Phase 1: write the log to real disk. */ + { + persist_t* p = wru_make_fresh_persist(tmp); + TEST_ASSERT_NOT_NULL(p); + + job_submit_data_t submit; + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)"q"; + submit.user = (char*)"tester"; + + xce_job_t jdef; + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = 73006; + jdef.state = JOB_STAT_RUN; + jdef.submit = &submit; + + uint8_t* job_payload = NULL; + size_t job_plen = 0; + TEST_ASSERT_EQUAL_INT(0, + wru_encode_job(&jdef, &job_payload, &job_plen)); + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_JOB_SUBMIT, job_payload, + (uint32_t)job_plen)); + free(job_payload); + + xce_job_runusage_t r = wru_make_test_runusage(); + uint8_t* rusage_payload = NULL; + size_t rusage_plen = 0; + TEST_ASSERT_EQUAL_INT(0, + wru_build_rusage_payload(73006, JOB_STAT_RUN, &r, + &rusage_payload, &rusage_plen)); + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_JOB_RUSAGE, rusage_payload, + (uint32_t)rusage_plen)); + free(rusage_payload); + + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + wru_teardown_persist(p); + } + + /* Phase 2: fresh persist_t + fresh cluster_t, recover from disk — + * simulates a controller restart. */ + { + persist_t* p2 = wru_make_fresh_persist(tmp); + TEST_ASSERT_NOT_NULL(p2); + + wru_setup(); + p2->cluster = &g_wru_c; + + int rc = persist_recover(p2); + TEST_ASSERT_EQUAL_INT(0, rc); + + job_t* j = wru_find_job(73006); + TEST_ASSERT_NOT_NULL_MESSAGE(j, + "restart replay must construct the job via WAL_JOB_SUBMIT"); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); + wru_assert_runusage_matches(j->job.run_usage); + + wru_teardown(); + wru_teardown_persist(p2); + } + + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp); + if (system(cmd) != 0) + fprintf(stderr, "cleanup warning: rm -rf %s failed\n", tmp); + + printf("test_wru_restart_replay_recovers_runusage: PASS\n"); +} + +/* Variadic trampoline so the test passes typed args through a real va_list + * -- pattern from test_persist_hooks.c / test_commit_barrier_rollout.c's + * call_job_hook, reproduced here (file-local) since that one is static to + * its own TU. */ +static int wru_call_job_hook(int type, int sit, int detail, persist_t* p, ...) +{ + va_list ap; + int rc; + + va_start(ap, p); + rc = persist_job_hook_fun(type, sit, detail, p, ap); + va_end(ap); + return rc; +} + +/* bug-1690's primary regression test: drives the REAL, unmodified + * persist_job_hook_fun() -- the actual production hook subscriber this fix + * changed -- with a job_t whose run_usage is genuinely populated (see this + * file's header comment for why this goes around update_job_rusages() + * itself rather than through it), through a real WAL-write + + * persist_recover() restart cycle. Asserts run_usage survives the restart + * intact, not just state -- the audit's exact confirmed root cause (mirrors + * bug-1678/1687/1689's own acceptance criterion). */ +static void test_wru_real_hook_appends_and_survives_restart(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_wru_realhook_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + const uint64_t jobid = 73007; + + /* Phase 1: real cluster + real disk persist. Seed WAL_JOB_SUBMIT + * directly (so replay can reconstruct the job) AND a live in-memory job + * object with run_usage already populated (so the real + * persist_job_hook_fun() call site has something to encode) -- mirrors + * test_wjs_real_on_job_start_survives_restart's dual-seed pattern. */ + { + persist_config_t cfg; + + if (makeString((char*)"wru-realhook-probe") == NULL) + xstring_table_init(64); + if (!g_wru_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_wru_jobobj_ready = 1; + } + + cluster_t c; + kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, + .nslots = 32, .delta = 16 }; + memset(&c, 0, sizeof(c)); + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&c.hooksys, XCE_CONTROLLER_HOOK_MAX)); + c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + TEST_ASSERT_NOT_NULL(c.jobmgr); + + cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + persist_set_workdir(&c.persist, tmp, "ctrl-test"); + TEST_ASSERT_EQUAL_INT(0, persist_init(&c.persist, &cfg, NULL)); + TEST_ASSERT_EQUAL_INT(0, persist_start(&c.persist)); + + /* Durable WAL_JOB_SUBMIT seed. */ + { + job_submit_data_t wal_submit; + xce_job_t wal_jdef; + uint8_t* buf = NULL; + size_t sz = 0; + + memset(&wal_submit, 0, sizeof(wal_submit)); + wal_submit.queue = (char*)"oq"; + wal_submit.user = (char*)"tester"; + + memset(&wal_jdef, 0, sizeof(wal_jdef)); + wal_jdef.jobid = jobid; + wal_jdef.state = JOB_STAT_RUN; + wal_jdef.submit = &wal_submit; + + TEST_ASSERT_EQUAL_INT(0, + wru_encode_job(&wal_jdef, &buf, &sz)); + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(&c.persist, WAL_JOB_SUBMIT, buf, + (uint32_t)sz)); + free(buf); + } + + /* Live in-memory job object, with run_usage populated directly + * (standing in for update_host_runusage()'s real accumulation -- + * see this file's header comment on why that function can't be + * exercised for real in this binary), for persist_job_hook_fun() + * to find/read. */ + job_t* live_job; + { + job_submit_data_t submit; + xce_job_t jdef; + + memset(&submit, 0, sizeof(submit)); + submit.queue = (char*)"oq"; + submit.user = (char*)"tester"; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = JOB_STAT_RUN; + jdef.submit = &submit; + + live_job = makeJob(c.jobmgr, &jdef); + TEST_ASSERT_NOT_NULL(live_job); + } + + xce_job_runusage_t r = wru_make_test_runusage(); + live_job->job.run_usage = &r; + + xce_job_rusages_t rusages; + memset(&rusages, 0, sizeof(rusages)); + rusages.jobid = jobid; + + int rc = wru_call_job_hook(XCE_CONTROLLER_HOOK_JOB, HOOK_AFTER_EVENT, + HOOK_DETAIL_JOB_RUSAGE_UPDATE, &c.persist, + &c, &rusages); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Prevent the stack-local `r` from being double-freed by teardown + * below — this fixture's job_t doesn't own it. */ + live_job->job.run_usage = NULL; + + TEST_ASSERT_EQUAL_INT(0, persist_flush(&c.persist)); + + persist_stop(&c.persist); + persist_fini(&c.persist); + dereference(c.jobmgr); + xce_hook_fini(&c.hooksys); + } + + /* Phase 2: fresh persist_t + fresh cluster_t (jobmgr/hooksys only), + * recover from disk — simulates a controller restart. */ + { + persist_t* p2 = wru_make_fresh_persist(tmp); + cluster_t c2; + job_t* j2 = NULL; + int rc; + kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, + .nslots = 32, .delta = 16 }; + + TEST_ASSERT_NOT_NULL(p2); + + memset(&c2, 0, sizeof(c2)); + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&c2.hooksys, XCE_CONTROLLER_HOOK_MAX)); + c2.jobmgr = makeInstance(&xkvmgr_type, &int_params); + TEST_ASSERT_NOT_NULL(c2.jobmgr); + + p2->cluster = &c2; + rc = persist_recover(p2); + TEST_ASSERT_EQUAL_INT(0, rc); + + lock(c2.jobmgr, RTTI_LOCK_READ); + key_find(c2.jobmgr, (void*)jobid, (void**)&j2); + unlock(c2.jobmgr); + TEST_ASSERT_NOT_NULL_MESSAGE(j2, + "restart replay must reconstruct the job via WAL_JOB_SUBMIT"); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j2->job.state); + + /* bug-1690: restart replay must recover the client-reported + * run_usage, not just state -- the audit's exact confirmed root + * cause (mirrors bug-1678/1687/1689's own acceptance criterion for + * their own fields). */ + wru_assert_runusage_matches(j2->job.run_usage); + + xce_hook_fini(&c2.hooksys); + if (c2.jobmgr) dereference(c2.jobmgr); + wru_teardown_persist(p2); + } + + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp); + if (system(cmd) != 0) + fprintf(stderr, "cleanup warning: rm -rf %s failed\n", tmp); + + printf("test_wru_real_hook_appends_and_survives_restart: PASS\n"); +} + +/* ── Test group entry point ───────────────────────────────────────────── */ + +void test_wal_job_rusage_fact(void) +{ + UnitySetTestFile(__FILE__); + RUN_TEST(test_wru_apply_committed_entry_sets_runusage_no_state_change); + RUN_TEST(test_wru_apply_committed_entry_sets_runusage_forward_transition); + RUN_TEST(test_wru_apply_committed_entry_backward_compat_old_payload); + RUN_TEST(test_wru_apply_committed_entry_merges_without_freeing_live_host); + RUN_TEST(test_wru_apply_wal_entry_sets_runusage_no_hook); + RUN_TEST(test_wru_apply_wal_entry_backward_compat_old_payload); + RUN_TEST(test_wru_restart_replay_recovers_runusage); + RUN_TEST(test_wru_real_hook_appends_and_survives_restart); +} diff --git a/src/include/wal_types.h b/src/include/wal_types.h index e6a0fc1f..86e6316e 100644 --- a/src/include/wal_types.h +++ b/src/include/wal_types.h @@ -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 diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index dd756cd7..e4ac2875 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -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");