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

fix(controller): WAL_JOB_CKPT loses job.ckptinfo on replay (bug-1691)

Browse source 
Same class as bug-1678/1687/1689/1690: on_job_ckpt() sets job.ckptinfo
(a plain char*, xstring-backed via makeString()) but is purely hook-driven
-- persist_job_hook_fun's default: arm only ever captured {jobid, state}
for WAL_JOB_CKPT. Neither apply seam restored ckptinfo after a restart.

Simplest of the five field-parity fixes: ckptinfo is a bare string, so no
capnp struct is needed -- mirrors WAL_QUEUE_CTRL's existing NUL-terminated-
string-after-header wire shape instead. Adoption is unconditional (not
gated on a state transition), matching bug-1690's precedent: a checkpoint
report never changes job.state. restore_job_ckpt_state() releases the
prior value via dereference() (the correct RTTI verb for an xstring), not
free().

Found by an /xce-audit-invariants controller pass (field-complete apply
parity dimension) -- the last of five sibling gaps found together
(bug-1678, bug-1687, bug-1689, bug-1690, this one). 7 new tests in
test_wal_job_ckpt_fact.c; RED/GREEN independently verified (5 genuine
pre-fix failures, 2 backward-compat tests correctly passed unchanged).
Full controller suite: 1167 tests, 2 failures -- both the pre-existing,
unrelated bug-1561 overlayfs sandbox flake. No ASAN findings.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AgYRU3zDTqNGV5Jk8aiKLg
This commit is contained in:
Rongsong Shen 2026-08-25 12:29:10 +00:00
parent b2a8fe4ce9
commit 977607615f
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
7 changed files with 955 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

44
src/controller/job.c
View file

@ -1023,6 +1023,50 @@ int restore_job_rusage_state(cluster_t* c, uint64_t jobid, uint32_t state,
return 0;
}
/* bug-1691: WAL_JOB_CKPT'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 ckptinfo string (NULL for a state-only
* WAL_JOB_CKPT entry, e.g. one written before this fix, or one fired while
* job.ckptinfo was still NULL at write time) so a promoted standby recovers
* the checkpoint info the same way apply_committed_entry's WAL_JOB_CKPT
* branch already restores it on the primary's own boot replay. Absolute
* state overwrite, matching every other restore_job_*_state() sibling on
* this seam -- no transition-result guarding happens on standby catch-up.
*
* `ckptinfo` is a TRANSIENT, caller-owned decode buffer (persist.c's
* decode_ckptinfo()), NOT itself an xstring -- j->job.ckptinfo is
* xstring-backed (makeString(), xstring.c), so adoption always goes through
* makeString() (which copies its input into its own refcounted allocation)
* and the existing value is released via dereference() (the RTTI verb),
* never free()/FREEUP, mirroring on_job_ckpt()'s own
* dereference(j->job.ckptinfo) call. `ckptinfo` itself is always freed here
* before returning, regardless of whether adoption happened -- the value
* actually stored in j->job.ckptinfo is always a freshly-made makeString()
* copy, never the input pointer itself. */
int restore_job_ckpt_state(cluster_t* c, uint64_t jobid, uint32_t state,
char* ckptinfo)
{
job_t* j = NULL;
lock(c->jobmgr, RTTI_LOCK_READ);
key_find(c->jobmgr, (void*)jobid, (void**)&j);
unlock(c->jobmgr);
if (!j) {
FREEUP(ckptinfo);
return -ENOENT;
}
lock(j, RTTI_LOCK_WRITE);
j->job.state = (int)state;
if (ckptinfo != NULL) {
if (j->job.ckptinfo != NULL)
dereference(j->job.ckptinfo);
j->job.ckptinfo = makeString(ckptinfo);
}
unlock(j);
FREEUP(ckptinfo);
return 0;
}
static int get_nslots_in_alloc(xce_host_alloc_t* alloc)
{
int i;

1
src/controller/meson.build
View file

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

173
src/controller/persist.c
View file

@ -11,6 +11,7 @@
#include "xlist.h"
#include "cluster.h"
#include "commit.h" /* commit_submit RETAIN registration (barrier-rollout 1.2) */
#include "xstring.h" /* makeString() -- bug-1691 WAL_JOB_CKPT ckptinfo adopt */
#include <assert.h>
#include <stdbool.h>
@ -74,6 +75,15 @@ extern int restore_job_finish_state(cluster_t* c, uint64_t jobid,
extern int restore_job_rusage_state(cluster_t* c, uint64_t jobid,
uint32_t state,
xce_job_runusage_t* runusage);
/* restore_job_ckpt_state defined in job.c (bug-1691): WAL_JOB_CKPT's own
* sibling of restore_job_state() above, additionally taking ownership of an
* optional decoded ckptinfo string (NULL for a state-only WAL_JOB_CKPT entry
* -- either one written before this fix, or one fired while job.ckptinfo was
* still NULL) so the checkpoint info survives standby catch-up / replicated-
* apply the same way apply_committed_entry's WAL_JOB_CKPT branch already
* restores it on the primary's own boot replay. */
extern int restore_job_ckpt_state(cluster_t* c, uint64_t jobid,
uint32_t state, char* ckptinfo);
/* 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
@ -324,6 +334,39 @@ static job_exitinfo_t* decode_jobexitinfo(const uint8_t* payload, size_t len)
return info;
}
/* Decode a native (non-capnp) WAL_JOB_CKPT trailing string payload (bug-1691)
* into a freshly heap-allocated, NUL-terminated char*. job.ckptinfo (xce/
* types.h) is a plain char* -- NOT the richer xce_ckptinfo_t/CheckpointInfo
* capnp struct used elsewhere for a different purpose -- so this mirrors
* WAL_QUEUE_CTRL's own [header][name NUL-terminated] wire shape
* (persist_queue_hook_fun / apply_wal_entry's WAL_QUEUE_CTRL case) rather
* than inventing a capnp wrapper for a bare string. Bounds-checked against
* `len` (the WAL entry's own payload_len, per the WAL_QUEUE_CTRL precedent)
* rather than trusting the bytes are NUL-terminated -- returns NULL if no
* NUL exists within the bounded region (a defensively-rejected malformed/
* truncated payload), mirroring decode_jobexitinfo()'s short-payload guard
* above. Returns NULL on a NULL/empty payload too; caller owns the result
* (release via free()). The returned buffer is a TRANSIENT decode copy, not
* itself adopted into job.ckptinfo -- makeString() (xstring.c) copies its
* input into its own refcounted allocation, so callers pass this result to
* makeString() and then free it (see restore_job_ckpt_state()/job.c and
* apply_committed_entry()'s own WAL_JOB_CKPT branch below). */
static char* decode_ckptinfo(const uint8_t* payload, size_t len)
{
char* copy;
if (!payload || len == 0)
return NULL;
if (memchr(payload, '\0', len) == NULL)
return NULL; /* not NUL-terminated within the declared bound -- reject */
copy = (char*)malloc(len);
if (!copy)
return NULL;
memcpy(copy, payload, len);
return copy;
}
/* 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
@ -3703,7 +3746,6 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
case WAL_JOB_STOP:
case WAL_JOB_RESUME:
case WAL_JOB_UPDATE:
case WAL_JOB_CKPT:
case WAL_JOB_PREASON:
/* Payload: wal_jobstate_t {jobid, state} */
if (payload_len >= sizeof(wal_jobstate_t)) {
@ -3712,6 +3754,32 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn,
}
break;
case WAL_JOB_CKPT: {
/* bug-1691: split out of the shared group above -- mirrors
* WAL_JOB_RUSAGE's split just below (bug-1690). This is the one
* fact type in that (former) group whose payload can carry a
* trailing NUL-terminated ckptinfo string (persist_job_hook_fun's
* HOOK_DETAIL_JOB_CKPT payload shape) alongside the {jobid, state}
* header, so a promoted standby's catch-up replay recovers the
* checkpoint info too, not just state. A payload of exactly
* sizeof(wal_jobstate_t) (no trailing bytes -- any WAL_JOB_CKPT
* entry written before this fix, or fired while job.ckptinfo was
* still NULL at write time) decodes no ckptinfo, 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;
char* ckptinfo = NULL;
if (payload_len > sizeof(wal_jobstate_t)) {
ckptinfo = decode_ckptinfo(
(const uint8_t*)payload + sizeof(wal_jobstate_t),
payload_len - sizeof(wal_jobstate_t));
}
restore_job_ckpt_state(p->cluster, js->jobid, js->state,
ckptinfo);
}
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/
@ -4147,6 +4215,49 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn,
merge_job_runusage(&(j->job.run_usage), decoded);
}
/* bug-1691: WAL_JOB_CKPT mirrors WAL_JOB_RUSAGE's unconditional-
* adopt pattern immediately above -- see that block's comment for
* the full rationale. A checkpoint report never changes job.state
* (on_job_ckpt(), job.c, only ever sets j->job.ckptinfo -- js->state
* is simply the job's UNCHANGED current state, captured verbatim by
* persist_job_hook_fun's HOOK_DETAIL_JOB_CKPT case), so gating this
* adoption on result==JOB_TRANSITION_APPLIED the way START/FINISH/
* DISPATCH do below would make it dead code: apply_job_update_
* transition() sees the wire state == the job's current state and
* reports JOB_TRANSITION_NOOP_DUPLICATE every time, so the gated
* block below would never run. Decode/adopt unconditionally, before
* the transition-guarded block. A payload of exactly
* sizeof(wal_jobstate_t) (no trailing bytes -- any WAL_JOB_CKPT
* entry written before this fix, or one fired while job.ckptinfo
* was still NULL at write time) decodes no ckptinfo, matching the
* pre-fix behavior exactly.
*
* j->job.ckptinfo is xstring-backed (makeString(), xstring.c), NOT
* a plain malloc'd/capnp-decoded string -- release the prior value
* via dereference() (the RTTI verb), never free()/FREEUP, mirroring
* on_job_ckpt()'s own dereference(j->job.ckptinfo) call. Unlike
* WAL_JOB_RUSAGE's run_usage, a free-and-replace here is safe (no
* merge needed): dereference() is the correct RTTI release call
* regardless of whether j->job.ckptinfo happens to be the SAME live
* xstring the primary's own live apply pump re-applies moments
* after on_job_ckpt() already set it -- it decrements a refcount
* rather than raw free()ing a non-owned pointer, so re-applying the
* identical fact is a harmless extra alloc/dereference cycle, not a
* hazard. decode_ckptinfo()'s result is a transient decode buffer
* (makeString() copies its input into its own allocation), so it is
* always freed here after use, whether or not adoption happened. */
if (wal_type == WAL_JOB_CKPT && len > sizeof(wal_jobstate_t)) {
char* decoded = decode_ckptinfo(
(const uint8_t*)payload + sizeof(wal_jobstate_t),
len - sizeof(wal_jobstate_t));
if (decoded != NULL) {
if (j->job.ckptinfo != NULL)
dereference(j->job.ckptinfo);
j->job.ckptinfo = makeString(decoded);
free(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
@ -7390,6 +7501,13 @@ int persist_job_hook_fun(int type, int situation, int detail,
* them. */
uint8_t* rusage_buf = NULL;
size_t rusage_len = 0;
/* bug-1691: WAL_JOB_CKPT's own optional trailing NUL-terminated
* ckptinfo string -- populated only by the HOOK_DETAIL_JOB_CKPT 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* ckpt_buf = NULL;
size_t ckpt_len = 0;
switch (detail) {
case HOOK_DETAIL_JOB_START: {
/* args: cluster_t*, job_pidinfo_t*, job_t* */
@ -7458,8 +7576,41 @@ int persist_job_hook_fun(int type, int situation, int detail,
&rusage_len);
break;
}
case HOOK_DETAIL_JOB_CKPT: {
/* args: cluster_t*, job_t* -- same shape as the default arm below,
* split out to also capture j->job.ckptinfo (bug-1691: same
* silent-field-drop bug class as bug-1678/1687/1689/1690 --
* WAL_JOB_CKPT previously carried only {jobid, state}, so a restart
* replay left job.ckptinfo permanently NULL even though
* on_job_ckpt() (job.c) had just set it, under the same write lock
* this AFTER_EVENT fires inside). j->job.ckptinfo is a plain
* NUL-terminated char* -- xstring-backed (makeString(), xstring.c),
* NOT the richer xce_ckptinfo_t/CheckpointInfo capnp struct used
* elsewhere for a different purpose -- so no capnp wrapper is
* needed; append it directly as a NUL-terminated string trailing
* the header, mirroring WAL_QUEUE_CTRL's own [header][name
* NUL-terminated] wire shape (persist_queue_hook_fun below)
* exactly. NULL only in test doubles that never set job.ckptinfo --
* the real production caller (on_job_ckpt) always sets it via
* makeString() before this hook fires; a NULL ckptinfo falls back
* to the legacy state-only payload at this function's shared tail,
* unchanged. */
(void)va_arg(ap, cluster_t*);
job_t* j = va_arg(ap, job_t*);
js.jobid = j->job.jobid; js.state = (uint32_t)j->job.state;
if (j->job.ckptinfo != NULL) {
ckpt_len = strlen(j->job.ckptinfo) + 1;
ckpt_buf = malloc(ckpt_len);
if (ckpt_buf != NULL)
memcpy(ckpt_buf, j->job.ckptinfo, ckpt_len);
else
ckpt_len = 0; /* OOM -- fall back to the state-only payload */
}
break;
}
default: {
/* STOP, RESUME, UPDATE, CKPT, PREASON — args: cluster_t*, job_t* */
/* STOP, RESUME, UPDATE, PREASON — args: cluster_t*, job_t* */
(void)va_arg(ap, cluster_t*);
job_t* j = va_arg(ap, job_t*);
js.jobid = j->job.jobid; js.state = (uint32_t)j->job.state;
@ -7487,6 +7638,24 @@ int persist_job_hook_fun(int type, int situation, int detail,
free(payload);
return rc;
}
if (ckpt_buf != NULL) {
/* bug-1691: append the header + trailing ckptinfo bytes as one
* contiguous payload -- mirrors the rusage_buf construction just
* above (bug-1690) and persist_queue_hook_fun's own header+trailer
* construction for WAL_QUEUE_CTRL (this function's own file, just
* below). */
size_t total = sizeof(js) + ckpt_len;
uint8_t* payload = malloc(total);
int rc;
if (!payload) { free(ckpt_buf); return -ENOMEM; }
memcpy(payload, &js, sizeof(js));
memcpy(payload + sizeof(js), ckpt_buf, ckpt_len);
free(ckpt_buf);
rc = wal_hook_append_classified(p, wal_type, mode, payload,
(uint32_t)total);
free(payload);
return rc;
}
return wal_hook_append_classified(p, wal_type, mode, &js, sizeof(js));
}

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

@ -87,6 +87,7 @@ 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_wal_job_ckpt_fact(void);
extern void test_job_dispatch_fail_count(void);
extern void test_apply_reapply_idempotency(void);
extern void test_apply_correctness_verification(void);
@ -260,6 +261,7 @@ int main(int argc, char* argv[])
TEST(test_wal_job_start_fact);
TEST(test_wal_job_finish_fact);
TEST(test_wal_job_rusage_fact);
TEST(test_wal_job_ckpt_fact);
TEST(test_job_dispatch_fail_count);
TEST(test_apply_reapply_idempotency);
TEST(test_apply_correctness_verification);

682
src/controller/testsuite/test_wal_job_ckpt_fact.c Normal file
View file

@ -0,0 +1,682 @@
/* test_wal_job_ckpt_fact.c — bug-1691 regression suite
*
* Proves WAL_JOB_CKPT'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), bug-1689 (WAL_JOB_FINISH/job.exit_info+finish_time), and
* bug-1690 (WAL_JOB_RUSAGE/job.run_usage) -- now carries the worker/
* scheduler-reported checkpoint info string (job.ckptinfo, a plain char*)
* 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)
*
* job.ckptinfo (xce/types.h) is a plain NUL-terminated char* -- NOT the
* richer xce_ckptinfo_t/CheckpointInfo capnp struct used elsewhere for a
* different purpose (that struct is on_job_ckpt()'s OWN parameter type,
* used only to carry the incoming report; the FIELD this fix restores is
* job.ckptinfo itself). Being a plain string, NO capnp struct is needed at
* all -- the fix appends job.ckptinfo as a NUL-terminated string directly
* after the wal_jobstate_t header, mirroring WAL_QUEUE_CTRL's own
* [header][name NUL-terminated] wire shape exactly (persist_queue_hook_fun /
* apply_wal_entry's WAL_QUEUE_CTRL case), rather than DISPATCH/START/RUSAGE's
* capnp-wrapper approach or FINISH's fixed-size native trailer.
*
* IMPORTANT DIVERGENCE FROM WAL_JOB_START/FINISH/DISPATCH's regression
* suites, SAME AS bug-1690's WAL_JOB_RUSAGE: on_job_ckpt() (job.c) never
* changes job.state -- it only ever sets job.ckptinfo. persist_job_hook_fun's
* new HOOK_DETAIL_JOB_CKPT case captures js.state verbatim from the job's
* CURRENT (unchanged) state, so apply_job_update_transition() sees wire
* state == current state and reports JOB_TRANSITION_NOOP_DUPLICATE on every
* single real CKPT replay. Gating ckptinfo adoption on
* result==JOB_TRANSITION_APPLIED (the way START/FINISH/DISPATCH do) would
* therefore make it dead code for the realistic, only-ever-occurring case.
* The fix decodes/adopts ckptinfo UNCONDITIONALLY, before that gate.
* test_wjc_apply_committed_entry_sets_ckptinfo_no_state_change below is this
* suite's most important test -- it is the scenario a naive copy of the
* START/FINISH/DISPATCH pattern would have silently failed to cover.
*
* ALSO DIVERGES FROM bug-1690's WAL_JOB_RUSAGE fix in one respect: adoption
* here is a plain free-and-replace (dereference() the old value, adopt a
* fresh makeString()'d copy), NOT a merge. job.ckptinfo IS xstring-backed
* (makeString(), xstring.c) exactly like RUSAGE's host_name field was, and
* CAN be the job's own live value when the primary's own live apply pump
* re-applies a fact moments after on_job_ckpt() already set it -- but
* dereference() is the correct RTTI release verb for an xstring regardless
* of whether the pointer is "old" or coincidentally "the current live one"
* (it decrements a refcount rather than raw free()ing a non-owned pointer),
* so no merge-vs-replace hazard exists here the way it did for RUSAGE's
* capnp-generated free_JobRunusage_ptr() (a plain-malloc-oriented free
* function that would have corrupted the heap on an xstring pointer).
*
* Also proves backward compatibility: a WAL_JOB_CKPT 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.ckptinfo was still NULL at write time) still decodes cleanly, with
* ckptinfo left untouched exactly as before this fix -- no crash, no
* misread.
*
* Fixtures mirror test_wal_job_finish_fact.c's own conventions exactly:
* - wjc_setup/wjc_teardown: bare cluster_t (jobmgr + hooksys) + in-memory
* persist_t.
* - wjc_make_fresh_persist/wjc_teardown_persist: real disk-backed
* persist_t + persist_recover(), for restart-replay tests.
*
* test_wjc_real_on_job_ckpt_survives_restart drives the REAL, unmodified
* on_job_ckpt() (the actual worker/scheduler-CKPT receipt handler --
* cluster.c's NotifyData_data_ckptinfo dispatch's only callee) through a
* real WAL-write + persist_recover() restart cycle -- unlike bug-1690's
* WAL_JOB_RUSAGE suite, on_job_ckpt() has no misc.c-style stubbed dependency
* in this test binary, so it can be driven directly (mirrors
* test_wjf_real_on_job_finish_survives_restart's simpler on_job_finish()
* pattern, not RUSAGE's persist_job_hook_fun trampoline workaround).
*/
#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_wjc_jobobj_ready;
static cluster_t g_wjc_c;
static persist_t g_wjc_p;
/* ── Spy hook: counts AFTER_EVENT firings, ignores all payload args ─────── */
static int g_wjc_hook_fires;
static int wjc_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_wjc_hook_fires++;
return 0;
}
static void wjc_setup(void)
{
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32,
.delta = 16 };
if (makeString((char*)"wjc-probe") == NULL)
xstring_table_init(64);
if (!g_wjc_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_wjc_jobobj_ready = 1;
}
memset(&g_wjc_c, 0, sizeof(g_wjc_c));
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&g_wjc_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
g_wjc_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
TEST_ASSERT_NOT_NULL(g_wjc_c.jobmgr);
memset(&g_wjc_p, 0, sizeof(g_wjc_p));
g_wjc_p.cluster = &g_wjc_c;
g_wjc_hook_fires = 0;
}
static void wjc_teardown(void)
{
xce_hook_fini(&g_wjc_c.hooksys);
if (g_wjc_c.jobmgr)
dereference(g_wjc_c.jobmgr);
}
/* Look up a job_t wrapper by jobid (NULL if absent). */
static job_t* wjc_find_job(uint64_t jobid)
{
job_t* j = NULL;
lock(g_wjc_c.jobmgr, RTTI_LOCK_READ);
int found = key_find(g_wjc_c.jobmgr, (void*)jobid, (void**)&j);
unlock(g_wjc_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_finish_fact.c's wjf_seed_job / test_wal_job_rusage_fact.c's
* wru_seed_job. */
static job_submit_data_t g_wjc_seed_submit;
static job_t* wjc_seed_job(uint64_t jobid, uint32_t state)
{
xce_job_t jdef;
memset(&g_wjc_seed_submit, 0, sizeof(g_wjc_seed_submit));
g_wjc_seed_submit.queue = (char*)"q";
g_wjc_seed_submit.user = (char*)"tester";
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = state;
jdef.submit = &g_wjc_seed_submit;
return makeJob(g_wjc_c.jobmgr, &jdef);
}
/* Build a wal_jobstate_t header + trailing NUL-terminated ckptinfo string --
* persist_job_hook_fun's WAL_JOB_CKPT payload shape (bug-1691), mirroring
* WAL_QUEUE_CTRL's own [header][name NUL-terminated] wire shape. Caller owns
* *out. */
static void wjc_build_ckpt_payload(uint64_t jobid, uint32_t state,
const char* ckptinfo,
uint8_t** out, size_t* out_len)
{
wal_jobstate_t js = { .jobid = jobid, .state = state, ._pad = 0 };
size_t info_len = strlen(ckptinfo) + 1;
*out_len = sizeof(js) + info_len;
*out = malloc(*out_len);
TEST_ASSERT_NOT_NULL(*out);
memcpy(*out, &js, sizeof(js));
memcpy(*out + sizeof(js), ckptinfo, info_len);
}
static void wjc_assert_ckptinfo_matches(job_t* j, const char* expected)
{
TEST_ASSERT_NOT_NULL_MESSAGE(j->job.ckptinfo,
"a WAL_JOB_CKPT payload with a trailing ckptinfo string must set "
"job.ckptinfo (bug-1691), not just state");
TEST_ASSERT_EQUAL_STRING(expected, j->job.ckptinfo);
}
/* ── apply_committed_entry(): live-commit apply / boot replay dispatch ──── */
static void test_wjc_apply_committed_entry_sets_ckptinfo_no_state_change(void)
{
wjc_setup();
job_t* j = wjc_seed_job(74001, JOB_STAT_RUN);
TEST_ASSERT_NOT_NULL(j);
uint8_t* payload = NULL;
size_t plen = 0;
wjc_build_ckpt_payload(74001, JOB_STAT_RUN, "ckpt-blob-1", &payload,
&plen);
xce_add_hook(&g_wjc_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjc_spy_hook,
NULL);
int rc = apply_committed_entry(&g_wjc_p, WAL_JOB_CKPT, 300, payload,
(uint32_t)plen);
free(payload);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state);
wjc_assert_ckptinfo_matches(j, "ckpt-blob-1");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wjc_hook_fires,
"a same-state (duplicate) CKPT replay -- the realistic, only-ever-"
"occurring case, since on_job_ckpt() never changes job.state -- "
"must NOT fire a notification hook, unchanged from before this fix, "
"even though ckptinfo WAS adopted");
wjc_teardown();
printf("test_wjc_apply_committed_entry_sets_ckptinfo_no_state_change: "
"PASS\n");
}
/* Secondary scenario: a WAL_JOB_CKPT fact that DOES carry a forward state
* transition (mechanism-level test, mirroring how bug-1690's WAL_JOB_RUSAGE
* suite covers this same divergence). ckptinfo must be adopted here too, and
* the existing forward-transition hook-fire behavior must be unchanged. */
static void
test_wjc_apply_committed_entry_sets_ckptinfo_forward_transition(void)
{
wjc_setup();
job_t* j = wjc_seed_job(74002, JOB_STAT_PEND);
TEST_ASSERT_NOT_NULL(j);
uint8_t* payload = NULL;
size_t plen = 0;
wjc_build_ckpt_payload(74002, JOB_STAT_RUN, "ckpt-blob-2", &payload,
&plen);
xce_add_hook(&g_wjc_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjc_spy_hook,
NULL);
int rc = apply_committed_entry(&g_wjc_p, WAL_JOB_CKPT, 301, payload,
(uint32_t)plen);
free(payload);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state);
wjc_assert_ckptinfo_matches(j, "ckpt-blob-2");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_wjc_hook_fires,
"a forward CKPT-carried transition must still fire exactly one "
"notification, unchanged from before this fix");
wjc_teardown();
printf(
"test_wjc_apply_committed_entry_sets_ckptinfo_forward_transition: "
"PASS\n");
}
static void
test_wjc_apply_committed_entry_backward_compat_old_payload(void)
{
wjc_setup();
job_t* j = wjc_seed_job(74003, JOB_STAT_RUN);
TEST_ASSERT_NOT_NULL(j);
TEST_ASSERT_NULL_MESSAGE(j->job.ckptinfo,
"fixture must seed a job with ckptinfo still unset");
wal_jobstate_t js = { .jobid = 74003, .state = JOB_STAT_RUN, ._pad = 0 };
int rc = apply_committed_entry(&g_wjc_p, WAL_JOB_CKPT, 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.ckptinfo,
"an old-format (exactly sizeof(wal_jobstate_t)) WAL_JOB_CKPT "
"payload must leave ckptinfo NULL -- no crash, no misread of "
"absent trailing bytes");
wjc_teardown();
printf("test_wjc_apply_committed_entry_backward_compat_old_payload: "
"PASS\n");
}
/* ── apply_wal_entry(): standby catch-up / replicated-apply dispatch ────── */
static void test_wjc_apply_wal_entry_sets_ckptinfo_no_hook(void)
{
wjc_setup();
job_t* j = wjc_seed_job(74004, JOB_STAT_RUN);
TEST_ASSERT_NOT_NULL(j);
uint8_t* payload = NULL;
size_t plen = 0;
wjc_build_ckpt_payload(74004, JOB_STAT_RUN, "ckpt-blob-4", &payload,
&plen);
xce_add_hook(&g_wjc_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjc_spy_hook,
NULL);
int rc = apply_wal_entry(&g_wjc_p, WAL_JOB_CKPT, 303, payload,
(uint32_t)plen);
free(payload);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state);
wjc_assert_ckptinfo_matches(j, "ckpt-blob-4");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wjc_hook_fires,
"apply_wal_entry's catch-up dispatch must never fire a "
"notification hook");
wjc_teardown();
printf("test_wjc_apply_wal_entry_sets_ckptinfo_no_hook: PASS\n");
}
static void test_wjc_apply_wal_entry_backward_compat_old_payload(void)
{
wjc_setup();
job_t* j = wjc_seed_job(74005, JOB_STAT_RUN);
TEST_ASSERT_NOT_NULL(j);
TEST_ASSERT_NULL_MESSAGE(j->job.ckptinfo,
"fixture must seed a job with ckptinfo still unset");
wal_jobstate_t js = { .jobid = 74005, .state = JOB_STAT_RUN, ._pad = 0 };
int rc = apply_wal_entry(&g_wjc_p, WAL_JOB_CKPT, 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.ckptinfo,
"an old-format WAL_JOB_CKPT payload must leave ckptinfo NULL on "
"this seam too");
wjc_teardown();
printf("test_wjc_apply_wal_entry_backward_compat_old_payload: PASS\n");
}
/* ── Restart-replay Observables ──────────────────────────────────────── */
static persist_t* wjc_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 wjc_teardown_persist(persist_t* p)
{
persist_stop(p);
persist_fini(p);
free(p);
}
/* Encode helper mirroring test_wal_job_finish_fact.c's wjf_encode_job
* (capnp_encode_job is file-static in persist.c). */
static int wjc_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_CKPT (real ckptinfo string) 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_finish_fact.c's own restart-replay test. */
static void test_wjc_restart_replay_recovers_ckptinfo(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_wjc_replay_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
const char* wire_ckptinfo = "ckpt-blob-replay";
/* Phase 1: write the log to real disk. */
{
persist_t* p = wjc_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 = 74006;
jdef.state = JOB_STAT_RUN;
jdef.submit = &submit;
uint8_t* job_payload = NULL;
size_t job_plen = 0;
TEST_ASSERT_EQUAL_INT(0,
wjc_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);
uint8_t* ckpt_payload = NULL;
size_t ckpt_plen = 0;
wjc_build_ckpt_payload(74006, JOB_STAT_RUN, wire_ckptinfo,
&ckpt_payload, &ckpt_plen);
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(p, WAL_JOB_CKPT, ckpt_payload,
(uint32_t)ckpt_plen));
free(ckpt_payload);
TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
wjc_teardown_persist(p);
}
/* Phase 2: fresh persist_t + fresh cluster_t, recover from disk —
* simulates a controller restart. */
{
persist_t* p2 = wjc_make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p2);
wjc_setup();
p2->cluster = &g_wjc_c;
int rc = persist_recover(p2);
TEST_ASSERT_EQUAL_INT(0, rc);
job_t* j = wjc_find_job(74006);
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);
wjc_assert_ckptinfo_matches(j, wire_ckptinfo);
wjc_teardown();
wjc_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_wjc_restart_replay_recovers_ckptinfo: PASS\n");
}
/* bug-1691's primary regression test: drives the REAL production
* on_job_ckpt() path (the actual worker/scheduler-CKPT receipt handler --
* cluster.c's NotifyData_data_ckptinfo dispatch's only callee) with a real
* xce_ckptinfo_t, through a real WAL-write + persist_recover() restart cycle
* -- the exact technique test_wjf_real_on_job_finish_survives_restart
* (bug-1689) already established. Asserts the job's ckptinfo survives the
* restart intact, not just state -- the audit's exact confirmed root
* cause. */
static void test_wjc_real_on_job_ckpt_survives_restart(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_wjc_onckpt_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
const uint64_t jobid = 74007;
const char* real_ckptinfo_wire = "real-ckpt-info-from-on-job-ckpt";
/* 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 (so the real on_job_ckpt() call site has something to
* find/lock) -- mirrors test_wjf_real_on_job_finish_survives_restart's
* dual-seed pattern exactly. Then drive an actual on_job_ckpt() call
* and let IT append WAL_JOB_CKPT via persist_job_hook_fun -- this is
* the production code path under test, not a hand-built WAL entry. */
{
persist_config_t cfg;
if (makeString((char*)"wjc-onckpt-probe") == NULL)
xstring_table_init(64);
if (!g_wjc_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_wjc_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");
/* Unlike test_wjf_real_on_job_finish_survives_restart's identical-
* looking fixture (bug-1689, a DIRECT persist_submit_job_finish_
* fact() call site -- no hook registration needed, so that sibling
* test can pass NULL here), on_job_ckpt() (job.c) is hook-driven:
* it only ever calls xce_run_hooks(..., HOOK_DETAIL_JOB_CKPT, ...)
* and relies ENTIRELY on persist_job_hook_fun (this fix's actual
* changed code) being registered as a c.hooksys subscriber to reach
* the WAL at all. persist_start() only registers that hook `if
* (p->cluster != NULL)` -- passing &c (not NULL) here is required
* for the real production path to append anything. */
TEST_ASSERT_EQUAL_INT(0, persist_init(&c.persist, &cfg, &c));
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,
wjc_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, already RUN, for on_job_ckpt() to
* find/lock. */
{
job_submit_data_t submit;
xce_job_t jdef;
memset(&submit, 0, sizeof(submit));
submit.queue = (char*)"oq";
submit.user = (char*)"tester";
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = JOB_STAT_RUN;
jdef.submit = &submit;
TEST_ASSERT_NOT_NULL(makeJob(c.jobmgr, &jdef));
}
/* The real worker/scheduler-reported checkpoint report. `ep` is
* unused by on_job_ckpt()'s body -- NULL is safe, mirroring how
* production call sites pass whatever endpoint_t* the transport
* layer happened to receive it on, which this fact-recording path
* never dereferences. */
xce_ckptinfo_t ckptinfo = { .jobid = jobid,
.ckptinfo = (char*)real_ckptinfo_wire };
int rc = on_job_ckpt(&c, NULL, &ckptinfo);
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c.persist));
persist_stop(&c.persist);
persist_fini(&c.persist);
dereference(c.jobmgr);
xce_hook_fini(&c.hooksys);
}
/* Phase 2: fresh persist_t + fresh cluster_t (jobmgr/hooksys only),
* recover from disk — simulates a controller restart. */
{
persist_t* p2 = wjc_make_fresh_persist(tmp);
cluster_t c2;
job_t* j2 = NULL;
int rc;
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32, .delta = 16 };
TEST_ASSERT_NOT_NULL(p2);
memset(&c2, 0, sizeof(c2));
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&c2.hooksys, XCE_CONTROLLER_HOOK_MAX));
c2.jobmgr = makeInstance(&xkvmgr_type, &int_params);
TEST_ASSERT_NOT_NULL(c2.jobmgr);
p2->cluster = &c2;
rc = persist_recover(p2);
TEST_ASSERT_EQUAL_INT(0, rc);
lock(c2.jobmgr, RTTI_LOCK_READ);
key_find(c2.jobmgr, (void*)jobid, (void**)&j2);
unlock(c2.jobmgr);
TEST_ASSERT_NOT_NULL_MESSAGE(j2,
"restart replay must reconstruct the job via WAL_JOB_SUBMIT");
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j2->job.state,
"restart replay must recover the job's state, unchanged by the "
"real on_job_ckpt() call");
/* bug-1691: restart replay must recover the worker/scheduler-
* reported ckptinfo string -- the audit's exact confirmed root
* cause (mirrors bug-1678/1687/1689/1690's own acceptance
* criterion). */
wjc_assert_ckptinfo_matches(j2, real_ckptinfo_wire);
xce_hook_fini(&c2.hooksys);
if (c2.jobmgr) dereference(c2.jobmgr);
wjc_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_wjc_real_on_job_ckpt_survives_restart: PASS\n");
}
/* ── Test group entry point ───────────────────────────────────────────── */
void test_wal_job_ckpt_fact(void)
{
UnitySetTestFile(__FILE__);
RUN_TEST(test_wjc_apply_committed_entry_sets_ckptinfo_no_state_change);
RUN_TEST(test_wjc_apply_committed_entry_sets_ckptinfo_forward_transition);
RUN_TEST(test_wjc_apply_committed_entry_backward_compat_old_payload);
RUN_TEST(test_wjc_apply_wal_entry_sets_ckptinfo_no_hook);
RUN_TEST(test_wjc_apply_wal_entry_backward_compat_old_payload);
RUN_TEST(test_wjc_restart_replay_recovers_ckptinfo);
RUN_TEST(test_wjc_real_on_job_ckpt_survives_restart);
}