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");