diff --git a/.wolf/buglog.json b/.wolf/buglog.json index dcfae1a5..056dcd9b 100644 --- a/.wolf/buglog.json +++ b/.wolf/buglog.json @@ -18874,6 +18874,38 @@ ], "occurrences": 1, "last_seen": "2026-08-25T11:00:00Z" + }, + { + "id": "bug-1691", + "timestamp": "2026-08-25T12:00:00Z", + "error_message": "xce-audit-invariants finding: WAL_JOB_CKPT silently drops job.ckptinfo (the worker/scheduler-reported checkpoint info string) 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), and bug-1690 (WAL_JOB_RUSAGE/job.run_usage). on_job_ckpt() (job.c) sets job->job.ckptinfo via makeString(), but the WAL_JOB_CKPT fact -- purely hook-driven via persist_job_hook_fun's HOOK_DETAIL_JOB_CKPT arm (on_job_ckpt() itself never calls a persist_submit_job_*_fact() function directly, unlike DISPATCH/START/FINISH) -- only ever carried the shared wal_jobstate_t{jobid, state} payload. Neither apply seam restored ckptinfo -- it stayed NULL after restart. job->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 incoming-report parameter type).", + "file": "src/controller/job.c:2999 (on_job_ckpt), src/controller/job.c:1046 (restore_job_ckpt_state, new), src/controller/persist.c:354 (decode_ckptinfo, new), src/controller/persist.c:7579 (persist_job_hook_fun's HOOK_DETAIL_JOB_CKPT case, new), src/controller/persist.c:3757 (apply_wal_entry's WAL_JOB_CKPT case, split out), src/controller/persist.c:4147 (apply_committed_entry's WAL_JOB_CKPT unconditional-adopt block), src/include/wal_types.h (WAL_JOB_CKPT doc comment), src/controller/testsuite/test_wal_job_ckpt_fact.c (new)", + "root_cause": "persist_job_hook_fun()'s job-fact switch (persist.c) routed HOOK_DETAIL_JOB_CKPT through its generic `default:` arm (shared with STOP/RESUME/UPDATE/PREASON), which captures only {jobid, state} from the job_t* -- j->job.ckptinfo, freshly set by on_job_ckpt() moments earlier under the same write lock this AFTER_EVENT fires inside, was never read or encoded. Both apply seams -- apply_committed_entry()'s shared WAL_JOB_START/FINISH/.../DISPATCH fall-through group (primary boot replay AND live commit-advance apply) and apply_wal_entry()'s identical shared group -> restore_job_state() (job.c, state-only) for standby catch-up -- never restored ckptinfo, matching the exact same class of gap bug-1678/1687/1689/1690 already fixed for allocs/pidinfo+start_time/exit_info+finish_time/run_usage. Confirmed via CodeGraph source read that on_job_ckpt() is hook-driven (fires xce_run_hooks only, no direct persist_submit_job_*_fact() call site) -- a genuine divergence from DISPATCH/START/FINISH's direct-call-site pattern, closer to bug-1690's RUSAGE precedent (also purely hook-driven) -- verified against the real code before implementing, per this task's explicit instruction. Also confirmed on_job_ckpt() never changes job.state (only job.ckptinfo), so a naive copy of the START/FINISH/DISPATCH pattern (gating adoption on a forward JOB_TRANSITION_APPLIED) would have made the fix dead code on every real invocation, mirroring bug-1690's exact same discovery for WAL_JOB_RUSAGE.", + "fix": "FIXED (2026-08-25). This is the simplest of the five field-parity fixes -- job.ckptinfo is a plain string, so NO capnp struct was needed at all (unlike bug-1678/1687/1690's capnp wrappers, though it does share bug-1689's 'native, non-capnp trailer' style). Mirrors WAL_QUEUE_CTRL's own established [header][name NUL-terminated] wire shape (persist_queue_hook_fun / apply_wal_entry's WAL_QUEUE_CTRL case, read as the reference implementation before designing this fix) exactly: the ckptinfo string is appended as a NUL-terminated string directly after the wal_jobstate_t header, bounded by the WAL entry's own payload_len (no separate length prefix). persist.c gains decode_ckptinfo() (a native memcpy-with-NUL-bound-check decode, mirroring decode_jobexitinfo()'s bug-1689 native-trailer style) and a new HOOK_DETAIL_JOB_CKPT case in persist_job_hook_fun() -- split out of the generic `default:` arm -- that encodes j->job.ckptinfo (when non-NULL) as trailing bytes appended after the wal_jobstate_t header, mirroring the rusage_buf/rusage_len construction bug-1690 already added to this same function for WAL_JOB_RUSAGE (both are hook-driven, unlike DISPATCH/START/FINISH's direct persist_submit_job_*_fact() call sites). src/include/wal_types.h's WAL_JOB_CKPT enum value gains a doc comment documenting the new payload shape and backward-compat contract, mirroring every other WAL_JOB_* entry's existing convention. Both apply seams updated: apply_wal_entry() (standby catch-up) splits WAL_JOB_CKPT out of its shared state-only group into its own case (mirroring bug-1690's WAL_JOB_RUSAGE split), decoding trailing ckptinfo bytes via decode_ckptinfo() and calling new restore_job_ckpt_state() (job.c, restore_job_rusage_state()'s sibling). apply_committed_entry() stays in its existing shared job-fact group but -- mirroring bug-1690's WAL_JOB_RUSAGE design exactly, for the identical reason -- decodes/adopts ckptinfo UNCONDITIONALLY, before the JOB_TRANSITION_APPLIED gate: a checkpoint report never changes job.state, so gating adoption on a forward state transition would make it dead code for the realistic, only-ever-occurring case (every real WAL_JOB_CKPT fact hits JOB_TRANSITION_NOOP_DUPLICATE, since js.state is captured verbatim from the job's own unchanged current state). A payload of exactly sizeof(wal_jobstate_t) (any WAL_JOB_CKPT entry written before this fix, or one fired while job.ckptinfo was still NULL at write time) decodes no ckptinfo on both seams, matching pre-fix behavior exactly -- full backward compatibility.\n\nAdoption is a plain free-and-replace (dereference() the prior xstring value, adopt a fresh makeString()'d copy), NOT a merge -- this DIVERGES from bug-1690's WAL_JOB_RUSAGE fix, which required a merge specifically because a naive free-and-replace via the WRONG release function (the capnp-generated free_JobRunusage_ptr(), written for plain malloc'd fields) would free a non-malloc'd, RTTI-managed xstring pointer. job.ckptinfo IS xstring-backed (makeString(), xstring.c) exactly like RUSAGE's host_name 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() (the correct RTTI release verb, used here instead of free()/FREEUP) is safe regardless of whether the pointer is 'old' or coincidentally 'the current live one', since it decrements a refcount rather than raw free()ing a non-owned pointer. This distinction was verified by reading xstring.c's actual makeString()/dereference() implementation before choosing the release strategy, not assumed from the RUSAGE precedent.\n\nRegression tests: new src/controller/testsuite/test_wal_job_ckpt_fact.c (7 tests, mirroring test_wal_job_rusage_fact.c's 'unconditional decode independent of transition result' structure plus test_wal_job_finish_fact.c's simpler real-production-call pattern, wired into src/controller/meson.build and testsuite/controller_test.c): (1) apply_committed_entry sets ckptinfo when state does NOT change (JOB_TRANSITION_NOOP_DUPLICATE) -- this suite's most important test, proving the unconditional-decode design choice, hook fires 0 times; (2) apply_committed_entry sets ckptinfo on a genuine forward transition, hook still fires once; (3) apply_committed_entry backward-compat with an old-format payload; (4) apply_wal_entry sets ckptinfo, fires no hook; (5) apply_wal_entry backward-compat; (6) direct-payload restart-replay (WAL_JOB_SUBMIT + new-format WAL_JOB_CKPT via persist_wal_async + persist_recover()); (7) test_wjc_real_on_job_ckpt_survives_restart -- drives the REAL production on_job_ckpt() path (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, asserting ckptinfo survives the restart intact -- the audit's exact confirmed root cause. Test 7 required passing a real cluster_t* (not NULL) as persist_init()'s 3rd argument -- unlike test_wjf_real_on_job_finish_survives_restart's identical-looking fixture (a direct-call-site fix needing no hook registration), on_job_ckpt() is hook-driven and depends entirely on persist_start() registering persist_job_hook_fun on cluster.hooksys (gated on `p->cluster != NULL`) to reach the WAL at all -- caught via a genuine RED-signal debugging pass (the real-path test initially failed even with the fix applied, isolating to this exact fixture gap, not a fix defect).\n\nRED->GREEN independently verified live (not just narrated): `git stash push -- src/controller/job.c src/controller/persist.c` (keeping the new test file, wal_types.h, meson.build/controller_test.c wiring), incremental rebuild via the synchronous xce-build.sh script, ran the full controller_unit_test suite -- 1167 Tests 7 Failures (the 2 pre-existing bug-1561 test_config_repl.c failures plus all 5 new-behavior WAL_JOB_CKPT tests failing with the exact expected 'must set job.ckptinfo' mismatch; the 2 backward-compat tests correctly PASS unchanged). `git stash pop`, incremental rebuild, ran the isolated new test group directly against the binary (`ASAN_OPTIONS=detect_leaks=0 ./controller_unit_test -t test_wal_job_ckpt_fact`, needed because LeakSanitizer's atexit-based leak check bypasses libc's normal exit() stdio-flush path on this binary, silently discarding all buffered PASS/FAIL output when leaks are present -- a pre-existing, suite-wide fixture characteristic unrelated to this fix, confirmed identical on the RUSAGE suite: 78 leaked allocations in isolation, same fixture-leak pattern, not a regression) -- initially 6/7 passed, with test_wjc_real_on_job_ckpt_survives_restart failing; root-caused to the persist_init() NULL-vs-&c fixture gap above, fixed the test, reran -- 7 Tests 0 Failures. Full controller_unit_test suite re-run via xce-build.sh after all fixes: 1167 Tests, 2 Failures -- exact bug-1561 signature match ('Expected 2 Was 3' / test_conf_trim_keeps_only_n_versions, 'Expected 0 Was ' / test_conf_get_saved_tag_no_config in test_config_repl.c), zero new failures, zero ASAN heap-corruption findings (grepped the full test log for heap-buffer-overflow/invalid-free/use-after-free/SEGV/Aborted/double-free -- none found).\n\nENVIRONMENT NOTE: this session's build/test runs were repeatedly disrupted by the /tmp filesystem filling to 100% (0 bytes free), traced to ~15,437 orphaned mkdtemp() scratch directories (xce_ckpt_*/xce_hook_test_*/xce_e2e_wal_*/etc., ~658GB total) accumulated across many prior sessions' test runs whose cleanup never ran (each interrupted mid-flight by the same ENOSPC condition in a compounding cycle). Cleared via `rm -rf /tmp/xce_*` (verified via /proc/*/fd that no live process held any of them open) before the final, valid GREEN run -- unrelated to this fix's correctness, but a genuine environment hazard worth flagging for other sessions sharing this machine.", + "tags": [ + "controller", + "job.c", + "persist.c", + "wal", + "wal-job-ckpt", + "job-ckptinfo", + "restart-recovery", + "data-loss", + "apply-seam", + "native-trailing-payload", + "hook-driven-fact", + "xstring-release-verb", + "test-fixture-hook-registration-gap" + ], + "related_bugs": [ + "bug-1678", + "bug-1687", + "bug-1689", + "bug-1690", + "bug-1561" + ], + "occurrences": 1, + "last_seen": "2026-08-25T12:20:00Z" } ] } diff --git a/src/controller/job.c b/src/controller/job.c index 3778e456..77c7b9a3 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -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; diff --git a/src/controller/meson.build b/src/controller/meson.build index f140c251..8e67298f 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -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', diff --git a/src/controller/persist.c b/src/controller/persist.c index 5dd90a83..e43dd713 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -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 #include @@ -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)); } diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index e4fd4ebb..b12d4980 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -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); diff --git a/src/controller/testsuite/test_wal_job_ckpt_fact.c b/src/controller/testsuite/test_wal_job_ckpt_fact.c new file mode 100644 index 00000000..1637fee6 --- /dev/null +++ b/src/controller/testsuite/test_wal_job_ckpt_fact.c @@ -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 + +#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_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); +} diff --git a/src/include/wal_types.h b/src/include/wal_types.h index 86e6316e..dc1a874f 100644 --- a/src/include/wal_types.h +++ b/src/include/wal_types.h @@ -164,7 +164,29 @@ typedef enum { * wrote -- freeing it via the capnp-generated * free_JobRunusage_ptr() would corrupt the * heap. */ - WAL_JOB_CKPT = 9, + WAL_JOB_CKPT = 9, /* worker/scheduler-reported checkpoint info update. + * Payload is wal_jobstate_t{jobid, state, _pad=0} + * for any entry written before bug-1691, or one + * fired while job.ckptinfo was still NULL at write + * time; as of bug-1691 persist_job_hook_fun's + * HOOK_DETAIL_JOB_CKPT case appends a trailing + * NUL-terminated ckptinfo string whenever + * job.ckptinfo 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. job.ckptinfo (xce/types.h) is a + * plain char* -- NOT the richer xce_ckptinfo_t/ + * CheckpointInfo capnp struct used elsewhere for a + * different purpose -- and is xstring-backed + * (makeString(), xstring.c), so this mirrors + * WAL_QUEUE_CTRL's own [header][name + * NUL-terminated] wire shape rather than inventing + * a capnp wrapper for a bare string. Like + * WAL_JOB_RUSAGE, adoption on the + * apply_committed_entry() seam is NOT gated on a + * forward state transition -- a checkpoint report + * never changes job.state, so gating it the same + * way would make it dead code. */ WAL_JOB_PREASON = 10, WAL_JOB_ARRAY_SUBMIT = 11, /* job-array-single-wal-record: one record for * the whole array (base spec + index range),