lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/src/controller/testsuite/test_wal_job_rusage_fact.c
Rongsong Shen 6df644e6d5
feat(controller): compute the LSF CPU peak and efficiency fields (bug-1736) 
update_job_cpu_stats() (misc.c), run after each usage report:
- each host's cpu_rate = CPU-seconds gained / wall seconds since its
  previous report; the job's cores = the sum over hosts
- peak_cpu = highest job cores seen (peak_cpu_time = it rounded, the
  wire field is an integer); peak_cpu_duration = the interval that set it
- avg_cpu_efficiency = cpu_time / elapsed / slots, peak_cpu_efficiency =
  peak_cpu / slots, mem_efficiency = highest used_mem / rusage[mem]
  request (job-wide total, per task by default), all in percent
aggregate_job_runusage() no longer treats cumulative cpu_time as a peak.

New fields host sample_time/cpu_rate and job peak_cpu are appended to
the capnp structs (old WAL entries still decode) and copied by
merge_job_runusage(), so a promoted standby keeps the rate baselines.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-10-02 04:14:35 +00:00

925 lines
35 KiB
C
Raw Blame History

/* test_wal_job_rusage_fact.c — bug-1690 regression suite
*
* Proves WAL_JOB_RUSAGE's payload -- previously bare wal_jobstate_t{jobid,
* state}, exactly the same silent-field-drop bug class as bug-1678
* (WAL_JOB_DISPATCH/job.allocs), bug-1687 (WAL_JOB_START/job.pidinfo+
* start_time) and bug-1689 (WAL_JOB_FINISH/job.exit_info+finish_time) --
* now carries the client-reported, accumulated per-host CPU/mem usage
* (job.run_usage, an xce_job_runusage_t*) through BOTH of persist.c's apply
* seams:
*
* - apply_committed_entry() (live-commit apply / boot replay dispatch)
* - apply_wal_entry() (standby catch-up / replicated-apply dispatch)
*
* SECOND, EQUALLY IMPORTANT DIVERGENCE: adoption is a MERGE
* (merge_job_runusage(), job.c), never a free-and-replace. job.run_usage
* can be the job's LIVE structure -- update_job_rusages()/update_host_
* runusage() (job.c/misc.c) mutate it synchronously and in place, with a
* host_name sourced from makeString() (an RTTI-managed, header-before-data
* pointer, NOT a plain malloc()) -- and the primary's own live apply pump
* (persist_apply_pump_run()) re-applies that SAME fact, on that SAME live
* job, moments after the live mutation already happened. Freeing it via
* the capnp-generated free_JobRunusage_ptr() (written for plain malloc'd,
* capnp-decoded Text fields) would free a non-malloc'd pointer -- heap
* corruption under ASAN. test_wru_apply_committed_entry_merges_without_
* freeing_live_host below proves this directly with a real makeString()'d
* host_name.
*
* IMPORTANT DIVERGENCE FROM WAL_JOB_START/FINISH/DISPATCH's regression
* suites: a rusage report almost never changes job.state (the job stays RUN
* across every repeat report) -- unlike a START/FINISH/DISPATCH, which are
* genuine one-time PEND->RUN / RUN->terminal transitions. apply_committed_
* entry()'s shared job-fact group only stamps type-specific fields when
* apply_job_update_transition() reports JOB_TRANSITION_APPLIED (a forward
* state change) -- gating run_usage adoption on that same condition would
* make it dead code for the realistic, common case (every report after the
* first hits JOB_TRANSITION_NOOP_DUPLICATE). The fix decodes/adopts
* run_usage UNCONDITIONALLY, before that gate. test_wru_apply_committed_
* entry_sets_runusage_no_state_change below is this suite's most important
* test: it proves run_usage still lands even when state does NOT change --
* the scenario a naive copy of the START/FINISH/DISPATCH pattern would have
* silently failed to cover.
*
* xce_job_runusage_t (xce/types.h) had NO existing Cap'n Proto mapping at
* all -- unlike JobAlloc/JobPIDInfo, which bug-1678/1687 could reuse
* standalone from the existing Job struct's own sub-fields. This fix adds a
* genuine new schema pair, HostRunusage + JobRunusage (controller.capnp),
* distinct from the similar-looking but unrelated xce_job_rusages_t/
* JobRusages/SchedJobRusages (the client's WIRE UPDATE REQUEST shape).
*
* Also proves backward compatibility: a WAL_JOB_RUSAGE entry with the OLD
* payload shape (exactly sizeof(wal_jobstate_t), no trailing bytes --
* simulating an entry written before this fix, or one fired while
* job.run_usage was still NULL at write time) still decodes cleanly, with
* run_usage left untouched exactly as before this fix -- no crash, no
* misread.
*
* Fixtures mirror test_wal_job_start_fact.c's own conventions exactly:
* - wru_setup/wru_teardown: bare cluster_t (jobmgr + hooksys) + in-memory
* persist_t.
* - wru_make_fresh_persist/wru_teardown_persist: real disk-backed
* persist_t + persist_recover(), for restart-replay tests.
*
* test_wru_real_hook_appends_and_survives_restart drives the REAL,
* unmodified persist_job_hook_fun() (the actual production hook subscriber
* this fix changed) via the "variadic trampoline" pattern established by
* test_commit_barrier_rollout.c/test_persist_hooks.c, rather than going
* through update_job_rusages() (job.c) itself: this same controller_unit_
* test binary already links test_commit_barrier_reply.c, which defines its
* OWN non-static stub of update_host_runusage() (a no-op, since misc.c is
* not part of controller_unit_test_src) -- calling update_job_rusages()
* from this file would silently hit that no-op stub and never populate
* job.run_usage at all, testing nothing new. Driving persist_job_hook_fun()
* directly with a job_t whose run_usage is set on the fixture (bypassing
* update_host_runusage entirely, which this fix does not touch) exercises
* the actual changed function on the actual production code path a real
* update_job_rusages() call reaches it from.
*/
#include <rtti.h>
#include <capnp_c.h>
#include <errno.h>
#include <stdarg.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "cluster.h"
#include "controller.capnp.h"
#include "hook.h"
#include "jobj.h"
#include "persist.h"
#include "wal_types.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xcealloc.h"
#include "xstring.h"
/* jobj.c's jobsub table and the process-global xstring table are shared
* across every test file in the controller_unit_test binary — guard
* idempotently, mirroring every sibling fixture in this suite. */
static int g_wru_jobobj_ready;
static cluster_t g_wru_c;
static persist_t g_wru_p;
/* ── Spy hook: counts AFTER_EVENT firings, ignores all payload args ─────── */
static int g_wru_hook_fires;
static int wru_spy_hook(int type, int situation, int detail, void* data,
va_list ap)
{
(void)type; (void)detail; (void)data; (void)ap;
if (situation == HOOK_AFTER_EVENT)
g_wru_hook_fires++;
return 0;
}
static void wru_setup(void)
{
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32,
.delta = 16 };
if (makeString((char*)"wru-probe") == NULL)
xstring_table_init(64);
if (!g_wru_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_wru_jobobj_ready = 1;
}
memset(&g_wru_c, 0, sizeof(g_wru_c));
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&g_wru_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
g_wru_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
TEST_ASSERT_NOT_NULL(g_wru_c.jobmgr);
memset(&g_wru_p, 0, sizeof(g_wru_p));
g_wru_p.cluster = &g_wru_c;
g_wru_hook_fires = 0;
}
static void wru_teardown(void)
{
xce_hook_fini(&g_wru_c.hooksys);
if (g_wru_c.jobmgr)
dereference(g_wru_c.jobmgr);
}
/* Look up a job_t wrapper by jobid (NULL if absent). */
static job_t* wru_find_job(uint64_t jobid)
{
job_t* j = NULL;
lock(g_wru_c.jobmgr, RTTI_LOCK_READ);
int found = key_find(g_wru_c.jobmgr, (void*)jobid, (void**)&j);
unlock(g_wru_c.jobmgr);
return (found > 0) ? j : NULL;
}
/* Creates a job directly in the fixture's jobmgr, bypassing dispatch/replay
* entirely — for seeding fact-transition tests, mirroring
* test_wal_job_dispatch_fact.c's wjd_seed_job / test_wal_job_start_fact.c's
* wjs_seed_job. */
static job_submit_data_t g_wru_seed_submit;
static job_t* wru_seed_job(uint64_t jobid, uint32_t state)
{
xce_job_t jdef;
memset(&g_wru_seed_submit, 0, sizeof(g_wru_seed_submit));
g_wru_seed_submit.queue = (char*)"q";
g_wru_seed_submit.user = (char*)"tester";
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = state;
jdef.submit = &g_wru_seed_submit;
return makeJob(g_wru_c.jobmgr, &jdef);
}
/* ── Real fixture data: 2 hosts' worth of accumulated CPU/mem usage ──────── */
static xce_host_runusage_t g_wru_host_a = {
.host_name = (char*)"wru-host-a", .cpu_time = 1200, .mem_usage = 512.5,
.sample_time = 1790000000, .cpu_rate = 1.25f /* bug-1736 */
};
static xce_host_runusage_t g_wru_host_b = {
.host_name = (char*)"wru-host-b", .cpu_time = 900, .mem_usage = 256.25,
.sample_time = 1790000030, .cpu_rate = 0.5f
};
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.peak_cpu = 1.75f; /* bug-1736 */
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_FLOAT(1.75f, got->peak_cpu); /* bug-1736 */
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_TRUE(got->host_runusage[0]->sample_time == 1790000000);
TEST_ASSERT_EQUAL_FLOAT(1.25f, got->host_runusage[0]->cpu_rate);
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);
TEST_ASSERT_TRUE(got->host_runusage[1]->sample_time == 1790000030);
TEST_ASSERT_EQUAL_FLOAT(0.5f, got->host_runusage[1]->cpu_rate);
}
/* 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, .sample_time = 1790000100, .cpu_rate = 2.5f
};
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.peak_cpu = 3.25f; /* bug-1736 */
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_FLOAT(3.25f, got->peak_cpu); /* bug-1736 */
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);
/* bug-1736: a merged existing host keeps the rate baseline */
TEST_ASSERT_TRUE(h->sample_time == 1790000100);
TEST_ASSERT_EQUAL_FLOAT(2.5f, h->cpu_rate);
} 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);
}
Reference in a new issue View git blame Copy permalink