From d0e5f85aa841f842044913f2d950e3117cceecf2 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Tue, 25 Aug 2026 13:14:28 +0000 Subject: [PATCH] fix(controller): WAL_JOB_PREASON loses job.pend_reasons on replay (bug-1692) Last of the five field-parity gaps from the original audit (bug-1678, bug-1687, bug-1689, bug-1690, bug-1691). set_preason()/clear_preason() set job.pend_reasons (host/resource-based pending-job explanations), but persist_job_hook_fun's default: arm only ever captured {jobid, state}. Neither apply seam restored pend_reasons after a restart. Reuses the ALREADY-EXISTING JobPendReason/HostBasedReason/ResourceBasedReason capnp structs -- no schema change needed, unlike bug-1690's WAL_JOB_RUSAGE. Safety-checked against bug-1690's lesson (RTTI/makeString()-backed fields need merge-not-replace, not free-and-replace): pend_reasons is plain malloc/strdup-backed throughout (erase_job_preason is FREEUP()-only, add_host_preason/add_res_preason use strdup()/calloc()), so a straightforward free-and-replace is safe here, matching bug-1678/1687's pattern instead. New wrinkle beyond the four prior fixes: pend_reasons has a genuine CLEAR operation, and "trailing bytes present or absent" can't distinguish a legitimate clear from an old-format payload -- added an explicit presence byte (0=cleared/adopt NULL, 1=capnp bytes follow), verified by two dedicated "clear survives restart" tests. 11 new tests in test_wal_job_preason_fact.c; RED/GREEN independently verified (7 genuine pre-fix failures, one test's own wrong premise about capnp's garbage-byte behavior caught and corrected during RED/GREEN iteration). Full controller suite: 1178 tests, 2 failures -- both the pre-existing, unrelated bug-1561 overlayfs sandbox flake. No ASAN findings. This closes the field-complete apply parity invariant class the xce-audit-invariants controller dimension 7 was added for. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01AgYRU3zDTqNGV5Jk8aiKLg --- .wolf/buglog.json | 34 + src/controller/job.c | 49 + src/controller/meson.build | 1 + src/controller/persist.c | 224 +++- src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_wal_job_preason_fact.c | 1120 +++++++++++++++++ src/include/wal_types.h | 38 +- 7 files changed, 1465 insertions(+), 3 deletions(-) create mode 100644 src/controller/testsuite/test_wal_job_preason_fact.c diff --git a/.wolf/buglog.json b/.wolf/buglog.json index 056dcd9b..d2951bcd 100644 --- a/.wolf/buglog.json +++ b/.wolf/buglog.json @@ -18906,6 +18906,40 @@ ], "occurrences": 1, "last_seen": "2026-08-25T12:20:00Z" + }, + { + "id": "bug-1692", + "timestamp": "2026-08-25T13:10:00Z", + "error_message": "xce-audit-invariants finding: WAL_JOB_PREASON silently drops job.pend_reasons (the scheduler-reported host/resource-based pending-job explanation) 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), bug-1690 (WAL_JOB_RUSAGE/job.run_usage), and bug-1691 (WAL_JOB_CKPT/job.ckptinfo). set_preason()/do_clr_preason() (job.c) set/clear job->pend_reasons (xce_job_preason_t*: main_reason plus xce_host_reason_t**/xce_res_reason_t** nested arrays), but the WAL_JOB_PREASON fact -- purely hook-driven via persist_job_hook_fun's `default:` arm (shared with STOP/RESUME/UPDATE, same class as bug-1690/1691's hook-driven precedent, NOT a direct persist_submit_job_*_fact() call site like DISPATCH/START/FINISH) -- only ever carried the shared wal_jobstate_t{jobid, state} payload. Neither apply seam restored pend_reasons after a restart. This was the LAST of the five field-parity gaps from the original xce-audit-invariants pass -- closes that invariant class.", + "file": "src/controller/persist.c:206 (capnp_encode_jobpreason, new), src/controller/persist.c:290 (decode_jobpreason, new), src/controller/persist.c:78 (persist_job_hook_fun local vars pend_buf/pend_len/pend_present), src/controller/persist.c (persist_job_hook_fun's new HOOK_DETAIL_SETPREASON/HOOK_DETAIL_CLRPREASON case, split out of default arm), src/controller/persist.c (persist_job_hook_fun's shared-tail pend_present append block), src/controller/persist.c (apply_wal_entry's WAL_JOB_PREASON case, split out of the WAL_JOB_FAIL/STOP/RESUME/UPDATE shared group), src/controller/persist.c (apply_committed_entry's WAL_JOB_PREASON unconditional-adopt block, added after the WAL_JOB_CKPT block, before the JOB_TRANSITION_APPLIED gate), src/controller/job.c (restore_job_preason_state, new, added after restore_job_ckpt_state), src/include/wal_types.h (WAL_JOB_PREASON=10 doc comment, new), src/controller/testsuite/test_wal_job_preason_fact.c (new, 11 tests), src/controller/meson.build, src/controller/testsuite/controller_test.c (test wiring)", + "root_cause": "persist_job_hook_fun()'s job-fact switch (persist.c) routed HOOK_DETAIL_SETPREASON/HOOK_DETAIL_CLRPREASON through its generic `default:` arm (shared with STOP/RESUME/UPDATE), which captures only {jobid, state} from the job_t* -- j->job.pend_reasons, freshly set/cleared by set_preason()/do_clr_preason() 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 pend_reasons, matching the exact same class of gap bug-1678/1687/1689/1690/1691 already fixed for allocs/pidinfo+start_time/exit_info+finish_time/run_usage/ckptinfo. Confirmed via CodeGraph source read that set_preason()/do_clr_preason() are hook-driven (fire xce_run_hooks only, no direct persist_submit_job_*_fact() call site) -- closer to bug-1690/1691's RUSAGE/CKPT precedent than DISPATCH/START/FINISH's direct-call-site pattern -- verified against the real code before implementing. Also confirmed set_preason()/do_clr_preason() never change job.state (only job.pend_reasons), 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/1691's exact same discovery. UNIQUE to this bug (verified, not assumed from precedent): job.pend_reasons has a genuine CLEAR operation (do_clr_preason/HOOK_DETAIL_CLRPREASON) that legitimately resets it to NULL and must survive replay just as durably as the SET side -- bug-1690/1691's own wire convention ('trailing bytes present or absent' == 'field has a new value or is old-format, leave untouched') cannot distinguish that legitimate NULL from a pre-fix payload, so a blind copy of that convention would have silently reintroduced a stale-pend_reasons-after-clear bug on restart.", + "fix": "FIXED (2026-08-25). controller.capnp ALREADY had a matching struct -- `struct JobPendReason $C.mapname(\"xce_job_preason_t\")` with `mainReason`/`hostReasons: List(HostBasedReason) $C.maplistcount(\"nhost_reasons\")`/`resReasons: List(ResourceBasedReason) $C.maplistcount(\"nres_reasons\")`, plus sibling `HostBasedReason $C.mapname(\"xce_host_reason_t\")` and `ResourceBasedReason $C.mapname(\"xce_res_reason_t\")` -- verified field-for-field against types.h's real struct layouts, and ALREADY wired into the full Job struct's own encode/decode/free (Job.pendreason field, used by the existing checkpoint/snapshot path) -- confirmed via grep that encode_JobPendReason/decode_JobPendReason/free_JobPendReason/free_JobPendReason_ptr/new_JobPendReason/read_JobPendReason/write_JobPendReason all already exist in the generated controller.capnp.c/.h. NO schema change needed (like bug-1678/1687's JobAlloc/JobPIDInfo reuse, unlike bug-1690's WAL_JOB_RUSAGE which needed a brand-new schema struct).\n\nSAFETY CHECK (bug-1690's RUSAGE lesson, explicitly re-verified rather than assumed): read erase_job_preason() (src/lib/preason.c) -- releases every field via FREEUP() (plain free()), never dereference()/RTTI. Read add_host_preason()/add_res_preason() (same file) -- build host/resource reason entries via strdup()/calloc() throughout, never makeString(). Read the generated decode_HostBasedReason/decode_ResourceBasedReason (controller.capnp.c) -- use STRING_DUP (plain strdup) + calloc(), and free_HostBasedReason/free_ResourceBasedReason/free_JobPendReason/free_JobPendReason_ptr all use plain free(). Also confirmed jobj.c's job_init() ALREADY does a plain MOVE_PTR(dst->job.pend_reasons, src->pend_reasons) when constructing a job from a decoded Job (no merge, no makeString) and job_destroy() releases it via erase_job_preason() (FREEUP-only) -- an existing, working precedent for exactly this adopt pattern elsewhere in the codebase. CONCLUSION: unlike WAL_JOB_RUSAGE's host_name (xstring/RTTI-backed, required a merge to avoid corrupting the codebase's own live in-place-mutated structure), job.pend_reasons and its entire nested tree are plain calloc()/strdup()-backed with NO RTTI involvement -- a straightforward free-old-then-adopt-new via the capnp-generated free_JobPendReason_ptr() is safe, matching bug-1678/1687's JobAlloc/JobPIDInfo pattern, NOT bug-1690's merge-required pattern. set_preason() itself corroborates this: it does erase_job_preason()+FREEUP() then a plain MOVE_PTR of the caller's already-built xce_job_preason_t*, never mutating the SAME live structure in place the way update_host_runusage() does for run_usage -- so there is no live-object-reapplied-via-apply-pump hazard analogous to RUSAGE's here either.\n\npersist.c gains capnp_encode_jobpreason()/decode_jobpreason() (mirroring capnp_encode_joballoc()/decode_joballoc() exactly). persist_job_hook_fun() gains a new HOOK_DETAIL_SETPREASON/HOOK_DETAIL_CLRPREASON case (split out of the generic default arm, mirroring bug-1691's HOOK_DETAIL_JOB_CKPT split) that encodes j->job.pend_reasons when non-NULL.\n\nKEY DESIGN DIVERGENCE from bug-1690/1691's wire convention (the CLEAR-semantics gap identified above): the payload appends an explicit PRESENCE BYTE after the wal_jobstate_t header whenever the fact is SETPREASON or CLRPREASON (not just when a value exists to encode) -- 0 = pend_reasons was genuinely cleared (adopt NULL on replay), 1 = trailing capnp JobPendReason bytes follow (adopt the decoded value). A payload of exactly sizeof(wal_jobstate_t) (no presence byte at all -- any entry written before this fix) leaves pend_reasons untouched on both apply seams, preserving full backward compatibility. Both apply seams updated: apply_wal_entry() splits WAL_JOB_PREASON out of its shared state-only group into its own case, decoding the presence byte (and trailing capnp bytes when present=1) via decode_jobpreason() and calling new restore_job_preason_state() (job.c) -- careful to treat a present-but-failed-decode as 'leave untouched', never as an implicit clear. apply_committed_entry() stays in its shared job-fact group but -- mirroring bug-1690/1691's unconditional-adopt design for the identical reason (set_preason()/do_clr_preason() never change job.state, so gating on JOB_TRANSITION_APPLIED would make it dead code) -- decodes/adopts pend_reasons UNCONDITIONALLY, before the transition gate, using the same presence-byte + decode-failure-safe logic.\n\nRegression tests: new src/controller/testsuite/test_wal_job_preason_fact.c (11 tests, mirroring test_wal_job_ckpt_fact.c's 'unconditional decode independent of transition result' structure, extended with PREASON-specific clear-semantics coverage no CKPT/RUSAGE precedent needed): (1) apply_committed_entry sets pend_reasons (BOTH a host-based AND a resource-based reason, exercising both nested list types) when state does NOT change -- most important test; (2) apply_committed_entry sets pend_reasons on a genuine forward transition; (3) apply_committed_entry backward-compat with an old bare payload; (4) apply_committed_entry CLEARS an existing pend_reasons on a presence-byte=0 payload -- the test that would have caught a naive copy of bug-1690/1691's convention; (5) apply_committed_entry leaves pend_reasons untouched when present=1 but zero trailing bytes follow (decode_jobpreason never even called) -- NOTE: an earlier draft of this test tried to exercise 'corrupt capnp bytes' instead and got a genuine RED signal for the wrong reason (Expected 4 Was 0) -- capnp-c's read path does not validate content, so garbage-but-nonempty bytes decode into a zero-valued struct rather than returning NULL (matching decode_joballoc/decode_jobrunusage's identical pre-existing leniency); corrected to the actually-reachable zero-trailing-bytes case; (6-8) the apply_wal_entry mirrors of (1)/(3)/(4); (9) direct-payload restart replay (WAL_JOB_SUBMIT + WAL_JOB_PREASON with host+resource reasons via persist_wal_async + persist_recover()); (10) test_wjp_real_do_set_preason_survives_restart -- drives the REAL, unmodified do_set_preason() (set_preason() itself is file-static to job.c; do_set_preason() is job.c's only exported entrypoint, matching the actual OPCode_setpreason wire path) through a real WAL-write + persist_recover() restart cycle, requiring a minimal zeroed (fd=-1, no iomgr) endpoint_t since do_set_preason() calls xce_reply() before processing -- mirrors test_commit_barrier_reply.c's established fixture pattern, packets drained via endpoint_deq()/free_packet() to avoid a reachable-leak false positive; (11) test_wjp_real_clear_survives_restart -- PREASON-specific, no CKPT/RUSAGE analog: drives real do_set_preason() THEN real do_clr_preason() before restarting, proving the CLEAR itself durably survives replay (not just the earlier SET) -- the audit's exact confirmed correctness requirement unique to this fact type.\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 doc comment, meson.build/controller_test.c wiring), incremental rebuild via the synchronous xce-build.sh script (`nix develop ./src --command bash src/tools/xce-build.sh controller asan build-asan --incremental`) -- 1178 Tests 9 Failures: the 2 pre-existing bug-1561 test_config_repl.c failures plus exactly 7 of the 11 new tests failing with the exact expected 'must set job.pend_reasons' / 'must reset pend_reasons to NULL' mismatches (tests 1,2,4,6,8,9,10 by suite order); the other 4 new tests (backward-compat x2, the decode-edge-case test, and the clear-survives-restart test) correctly PASS unchanged under the reverted code -- verified this is expected, not a test-quality defect: those 4 scenarios are either genuinely unaffected by the fix (backward-compat) or, for the two restart-replay tests, happen to trivially satisfy their own assertion under BOTH old and new code (old code never encodes pend_reasons at all regardless of set-vs-clear, so a 'must be NULL after clear' check passes vacuously pre-fix too -- acceptable, since the SET-side restart test (10) and the direct-payload test (9) already independently prove the encode/decode mechanism itself, and test (11)'s value is in proving the CLEAR case specifically once mechanism 10 is proven, not in isolating it alone). `git stash pop`, incremental rebuild -- 1178 Tests 2 Failures (only the pre-existing bug-1561 pair, exact signature match 'Expected 2 Was 3' / 'Expected 0 Was 11954236611186853435'), all 11 new tests pass, zero new failures, zero ASAN heap-corruption findings (grepped the full test log for AddressSanitizer/heap-buffer-overflow/invalid-free/double-free/use-after-free/SEGV/Aborted -- none found). Verified no new compiler warnings in persist.c/job.c/the new test file (isolated rebuild + warning grep; the handful of pre-existing format-truncation/comment warnings in persist.c are all at unrelated line numbers, predating this change).\n\nThis was the LAST of the five field-parity gaps identified by the original xce-audit-invariants pass (bug-1678/1687/1689/1690/1691/1692) -- that invariant class (APPLY_CLEAN's entity-count+cursor-position check not proving per-field completeness, persistence.md) is now closed for job.allocs/pidinfo+start_time/exit_info+finish_time/run_usage/ckptinfo/pend_reasons.", + "tags": [ + "controller", + "job.c", + "persist.c", + "wal", + "wal-job-preason", + "job-pend-reasons", + "restart-recovery", + "data-loss", + "apply-seam", + "capnp-schema-reuse", + "hook-driven-fact", + "clear-semantics", + "presence-byte-wire-shape", + "test-quality-correction" + ], + "related_bugs": [ + "bug-1678", + "bug-1687", + "bug-1689", + "bug-1690", + "bug-1691", + "bug-1561" + ], + "occurrences": 1, + "last_seen": "2026-08-25T13:10:00Z" } ] } diff --git a/src/controller/job.c b/src/controller/job.c index 77c7b9a3..eec860a7 100644 --- a/src/controller/job.c +++ b/src/controller/job.c @@ -1067,6 +1067,55 @@ int restore_job_ckpt_state(cluster_t* c, uint64_t jobid, uint32_t state, return 0; } +/* bug-1692: WAL_JOB_PREASON's own sibling of restore_job_ckpt_state() above, + * for apply_wal_entry()'s standby catch-up / replicated-apply seam. + * + * has_field distinguishes a pre-fix bare {jobid,state} payload (has_field=0, + * pend_reasons left untouched -- full backward compatibility) from a + * post-fix payload that explicitly carries pend_reasons' new value + * (has_field=1): preason non-NULL adopts a freshly-decoded value, preason + * NULL means the field was genuinely cleared (do_clr_preason) and + * pend_reasons should be reset to NULL too -- pend_reasons is the one + * field-parity fact type in this family (bug-1678/1687/1689/1690/1691) with + * a real CLEAR operation, so unlike restore_job_ckpt_state()/ + * restore_job_rusage_state() a NULL decoded value is NOT automatically + * "nothing to adopt". + * + * preason (when non-NULL) is always freshly heap-allocated by + * decode_jobpreason() (persist.c) -- plain calloc()/strdup() throughout + * (verified via erase_job_preason()'s FREEUP-only release and every + * add_*_preason() constructor in src/lib/preason.c, plus the + * capnp-generated decode_HostBasedReason/decode_ResourceBasedReason using + * STRING_DUP+calloc), NOT xstring/RTTI-backed like ckptinfo or rusage's + * host_name -- so a straightforward free-old-then-adopt-new via + * free_JobPendReason_ptr() is safe here, no merge-not-replace treatment + * needed (unlike bug-1690's WAL_JOB_RUSAGE). Ownership of preason always + * transfers to job->job.pend_reasons on success; released here if the job + * is gone. */ +int restore_job_preason_state(cluster_t* c, uint64_t jobid, uint32_t state, + int has_field, xce_job_preason_t* preason) +{ + job_t* j = NULL; + lock(c->jobmgr, RTTI_LOCK_READ); + key_find(c->jobmgr, (void*)jobid, (void**)&j); + unlock(c->jobmgr); + if (!j) { + if (preason != NULL) + free_JobPendReason_ptr(&preason); + return -ENOENT; + } + + lock(j, RTTI_LOCK_WRITE); + j->job.state = (int)state; + if (has_field) { + if (j->job.pend_reasons != NULL) + free_JobPendReason_ptr(&(j->job.pend_reasons)); + j->job.pend_reasons = preason; /* NULL (cleared) or the decoded value */ + } + unlock(j); + return 0; +} + static int get_nslots_in_alloc(xce_host_alloc_t* alloc) { int i; diff --git a/src/controller/meson.build b/src/controller/meson.build index 8e67298f..acd89271 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -174,6 +174,7 @@ controller_unit_test_src = [ 'testsuite/test_wal_job_finish_fact.c', 'testsuite/test_wal_job_rusage_fact.c', 'testsuite/test_wal_job_ckpt_fact.c', + 'testsuite/test_wal_job_preason_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 e43dd713..1334d8f7 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -84,6 +84,19 @@ extern int restore_job_rusage_state(cluster_t* c, uint64_t jobid, * 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); +/* restore_job_preason_state defined in job.c (bug-1692): WAL_JOB_PREASON's own + * sibling of restore_job_state() above. Unlike restore_job_ckpt_state()/ + * restore_job_rusage_state(), a NULL decoded pointer is not automatically + * "leave untouched" here -- pend_reasons has a genuine CLEAR operation + * (do_clr_preason/HOOK_DETAIL_CLRPREASON) that must durably survive replay + * too, so has_field disambiguates a pre-fix bare {jobid,state} payload + * (has_field=0, pend_reasons left untouched -- full backward compatibility) + * from a post-fix payload that explicitly carries the field's new value + * (has_field=1): preason non-NULL adopts a freshly-decoded value, preason + * NULL means pend_reasons was genuinely cleared and must become NULL too. */ +extern int restore_job_preason_state(cluster_t* c, uint64_t jobid, + uint32_t state, int has_field, + xce_job_preason_t* preason); /* 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 @@ -253,6 +266,67 @@ static xce_job_alloc_t* decode_joballoc(const uint8_t* payload, size_t len) return alloc; } +/* Serialize xce_job_preason_t to a malloc'd Cap'n Proto buffer (bug-1692): + * lets WAL_JOB_PREASON carry the scheduler's host/resource-based pend + * reasons alongside the {jobid, state} header, so a restart replay can + * reconstruct job.pend_reasons instead of leaving it permanently NULL. + * Mirrors capnp_encode_joballoc above exactly, using the JobPendReason + * struct (already in controller.capnp as the existing Job.pendreason field + * -- no schema change needed here). */ +static int capnp_encode_jobpreason(const xce_job_preason_t* pr, uint8_t** out, + size_t* out_len) +{ + struct capn c; + struct capn_segment* cs; + struct JobPendReason cjp; + JobPendReason_ptr jpp; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_JobPendReason(cs, &cjp, (xce_job_preason_t*)pr); + jpp = new_JobPendReason(cs); + write_JobPendReason(&cjp, jpp); + capn_setp(capn_root(&c), 0, jpp.p); + + *out_len = capn_size(&c); + *out = malloc(*out_len); + if (!*out) { capn_free(&c); return -ENOMEM; } + capn_write_mem(&c, *out, *out_len, 0); + capn_free(&c); + return 0; +} + +/* Decode a Cap'n Proto JobPendReason buffer (bug-1692) into a freshly heap- + * allocated xce_job_preason_t*, mirroring decode_joballoc()'s pattern above. + * Returns NULL on a NULL/empty payload; caller owns the result. The decoded + * tree is plain calloc()/strdup() throughout (capnp-generated + * decode_HostBasedReason/decode_ResourceBasedReason use STRING_DUP + calloc, + * verified in controller.capnp.c), NOT xstring/RTTI-backed like ckptinfo or + * rusage's host_name -- release via free_JobPendReason_ptr (plain free()), + * never dereference(). */ +static xce_job_preason_t* decode_jobpreason(const uint8_t* payload, + size_t len) +{ + struct capn c_ctx; + JobPendReason_ptr jpp; + struct JobPendReason cjp; + xce_job_preason_t* preason; + + if (!payload || len == 0) + return NULL; + + capn_init_mem(&c_ctx, payload, len, 0); + jpp.p = capn_getp(capn_root(&c_ctx), 0, 1); + memset(&cjp, 0, sizeof(cjp)); + read_JobPendReason(&cjp, jpp); + + preason = calloc(1, sizeof(*preason)); + if (!preason) { capn_free(&c_ctx); return NULL; } + decode_JobPendReason(preason, &cjp); + capn_free(&c_ctx); + return preason; +} + /* Serialize job_pidinfo_t to a malloc'd Cap'n Proto buffer (bug-1687): * lets WAL_JOB_START carry the worker-reported pids/pgids/exdata alongside * the {jobid, state} header, so a restart replay can reconstruct @@ -3746,7 +3820,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_PREASON: /* Payload: wal_jobstate_t {jobid, state} */ if (payload_len >= sizeof(wal_jobstate_t)) { const wal_jobstate_t* js = (const wal_jobstate_t*)payload; @@ -3754,6 +3827,46 @@ int apply_wal_entry(persist_t* p, uint32_t type, uint64_t lsn, } break; + case WAL_JOB_PREASON: { + /* bug-1692: split out of the shared group above -- mirrors + * WAL_JOB_CKPT's/WAL_JOB_RUSAGE's splits (bug-1690/1691). Unlike + * those, pend_reasons has a genuine "clear" operation + * (do_clr_preason), so the payload beyond the wal_jobstate_t header + * always starts with an explicit presence byte (0 = cleared, adopt + * NULL; 1 = capnp JobPendReason bytes follow, adopt the decoded + * value) on any entry written after this fix. A payload of exactly + * sizeof(wal_jobstate_t) (no presence byte -- any WAL_JOB_PREASON + * entry written before this fix) leaves pend_reasons untouched, + * 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; + int has_field = 0; + xce_job_preason_t* preason = NULL; + if (payload_len > sizeof(wal_jobstate_t)) { + uint8_t present = + *((const uint8_t*)payload + sizeof(wal_jobstate_t)); + if (present) { + if (payload_len > sizeof(wal_jobstate_t) + 1) { + preason = decode_jobpreason( + (const uint8_t*)payload + sizeof(wal_jobstate_t) + + 1, + payload_len - sizeof(wal_jobstate_t) - 1); + } + /* present-but-failed-decode (corrupt/truncated trailing + * bytes) must NOT be treated as a clear -- leave + * pend_reasons untouched, matching every other decode + * helper's failure fallback in this function. */ + has_field = (preason != NULL); + } else { + has_field = 1; /* explicit clear -- adopt NULL */ + } + } + restore_job_preason_state(p->cluster, js->jobid, js->state, + has_field, preason); + } + 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 @@ -4258,6 +4371,51 @@ int apply_committed_entry(persist_t* p, uint32_t wal_type, uint64_t lsn, } } + /* bug-1692: WAL_JOB_PREASON mirrors WAL_JOB_CKPT's/WAL_JOB_RUSAGE's + * unconditional-adopt pattern immediately above -- set_preason()/ + * do_clr_preason() (job.c) never touch job.state, so gating + * adoption on result==JOB_TRANSITION_APPLIED would make it dead + * code: js->state is always the job's unchanged current state + * (persist_job_hook_fun's HOOK_DETAIL_SETPREASON/CLRPREASON arm + * captures it verbatim), so apply_job_update_transition() always + * reports JOB_TRANSITION_NOOP_DUPLICATE here. Decode/adopt + * unconditionally, before the transition-guarded block below. + * + * Unlike CKPT/RUSAGE, a presence byte (not "trailing bytes exist or + * not") disambiguates a genuine CLEAR (do_clr_preason) from an + * old-format payload -- see decode_jobpreason()'s/ + * restore_job_preason_state()'s doc comments and WAL_JOB_PREASON's + * own doc comment in wal_types.h for the full rationale. `present == + * 0` (explicit clear) or a successful decode both trigger adoption; + * a present-but-failed decode (corrupt/truncated trailing bytes) + * leaves the existing value untouched, matching every other type's + * decode-failure fallback in this function. + * + * pend_reasons is plain calloc()/strdup()-backed throughout + * (verified via erase_job_preason()'s FREEUP-only release and every + * add_*_preason() constructor in src/lib/preason.c, plus the + * capnp-generated decode_HostBasedReason/decode_ResourceBasedReason + * using STRING_DUP+calloc), NOT xstring/RTTI-backed like + * ckptinfo/rusage's host_name -- so a straightforward + * free-and-replace via free_JobPendReason_ptr() is safe here, no + * merge-not-replace treatment needed (unlike bug-1690's + * WAL_JOB_RUSAGE). */ + if (wal_type == WAL_JOB_PREASON && len > sizeof(wal_jobstate_t)) { + uint8_t present = + *((const uint8_t*)payload + sizeof(wal_jobstate_t)); + xce_job_preason_t* decoded = NULL; + if (present && len > sizeof(wal_jobstate_t) + 1) { + decoded = decode_jobpreason( + (const uint8_t*)payload + sizeof(wal_jobstate_t) + 1, + len - sizeof(wal_jobstate_t) - 1); + } + if (present == 0 || decoded != NULL) { + if (j->job.pend_reasons != NULL) + free_JobPendReason_ptr(&(j->job.pend_reasons)); + j->job.pend_reasons = decoded; /* NULL (cleared) or 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 @@ -7508,6 +7666,19 @@ int persist_job_hook_fun(int type, int situation, int detail, * wal_jobstate_t it always was -- zero behavior change for them. */ uint8_t* ckpt_buf = NULL; size_t ckpt_len = 0; + /* bug-1692: WAL_JOB_PREASON's own optional trailing payload -- populated + * only by the HOOK_DETAIL_SETPREASON/HOOK_DETAIL_CLRPREASON arm below, + * consumed by the shared append at the bottom of this function. + * pend_present stays -1 for every OTHER detail (unchanged, plain + * wal_jobstate_t payload). Unlike rusage_buf/ckpt_buf above, + * pend_reasons has a genuine CLEAR operation (do_clr_preason) that must + * be distinguishable on replay from "old-format payload, leave + * untouched" -- so this fact type always appends an explicit presence + * byte (0 or 1) once pend_present >= 0, even when pend_buf stays NULL + * (the clear case). */ + uint8_t* pend_buf = NULL; + size_t pend_len = 0; + int pend_present = -1; switch (detail) { case HOOK_DETAIL_JOB_START: { /* args: cluster_t*, job_pidinfo_t*, job_t* */ @@ -7609,8 +7780,36 @@ int persist_job_hook_fun(int type, int situation, int detail, } break; } + case HOOK_DETAIL_SETPREASON: + case HOOK_DETAIL_CLRPREASON: { + /* args: cluster_t*, job_t* -- same shape as the default arm below, + * split out to also capture j->job.pend_reasons (bug-1692: same + * silent-field-drop bug class as bug-1678/1687/1689/1690/1691 -- + * WAL_JOB_PREASON previously carried only {jobid, state}, so a + * restart replay left job.pend_reasons permanently NULL/stale even + * though set_preason()/do_clr_preason() (job.c) had just set/cleared + * it, under the same write lock this AFTER_EVENT fires inside). + * pend_present distinguishes "cleared" (0, do_clr_preason has just + * FREEUP'd job.pend_reasons back to NULL) from "set" (1, capnp + * JobPendReason bytes follow) -- both HOOK_DETAIL_SETPREASON and + * HOOK_DETAIL_CLRPREASON share this arm since the shared append + * tail below decides purely from j->job.pend_reasons' CURRENT value, + * not from which detail fired. */ + (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.pend_reasons != NULL) { + pend_present = 1; + (void)capnp_encode_jobpreason(j->job.pend_reasons, &pend_buf, + &pend_len); + } else { + pend_present = 0; + } + break; + } default: { - /* STOP, RESUME, UPDATE, PREASON — args: cluster_t*, job_t* */ + /* STOP, RESUME, UPDATE — 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; @@ -7656,6 +7855,27 @@ int persist_job_hook_fun(int type, int situation, int detail, free(payload); return rc; } + if (pend_present >= 0) { + /* bug-1692: append [header][presence byte][optional capnp + * JobPendReason bytes] as one contiguous payload -- mirrors the + * rusage_buf/ckpt_buf constructions just above (bug-1690/1691), but + * ALWAYS appends (even when pend_buf is NULL, the clear case) since + * the presence byte itself is the signal both apply seams need to + * distinguish "cleared" from "old-format, don't touch". */ + size_t total = sizeof(js) + 1 + pend_len; + uint8_t* payload = malloc(total); + int rc; + if (!payload) { free(pend_buf); return -ENOMEM; } + memcpy(payload, &js, sizeof(js)); + payload[sizeof(js)] = (uint8_t)pend_present; + if (pend_len > 0) + memcpy(payload + sizeof(js) + 1, pend_buf, pend_len); + free(pend_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 b12d4980..2a5bd302 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -88,6 +88,7 @@ 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_wal_job_preason_fact(void); extern void test_job_dispatch_fail_count(void); extern void test_apply_reapply_idempotency(void); extern void test_apply_correctness_verification(void); @@ -262,6 +263,7 @@ int main(int argc, char* argv[]) TEST(test_wal_job_finish_fact); TEST(test_wal_job_rusage_fact); TEST(test_wal_job_ckpt_fact); + TEST(test_wal_job_preason_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_preason_fact.c b/src/controller/testsuite/test_wal_job_preason_fact.c new file mode 100644 index 00000000..75a3439d --- /dev/null +++ b/src/controller/testsuite/test_wal_job_preason_fact.c @@ -0,0 +1,1120 @@ +/* test_wal_job_preason_fact.c — bug-1692 regression suite + * + * Proves WAL_JOB_PREASON'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), bug-1690 + * (WAL_JOB_RUSAGE/job.run_usage), and bug-1691 (WAL_JOB_CKPT/job.ckptinfo) -- + * now carries the scheduler-reported host/resource-based pending reasons + * (job.pend_reasons, xce_job_preason_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) + * + * The JobPendReason/HostBasedReason/ResourceBasedReason Cap'n Proto structs + * ALREADY existed in controller.capnp (used elsewhere for the full-Job + * snapshot/checkpoint path) and matched xce_job_preason_t/xce_host_reason_t/ + * xce_res_reason_t field-for-field, so NO schema change was needed -- this + * fix reuses the already-generated encode_JobPendReason/decode_JobPendReason/ + * free_JobPendReason_ptr functions directly, mirroring bug-1678's + * WAL_JOB_DISPATCH/JobAlloc reuse pattern. + * + * IMPORTANT DIVERGENCE FROM WAL_JOB_START/FINISH/DISPATCH's regression + * suites, SAME AS bug-1690/1691: set_preason()/do_clr_preason() (job.c) + * never change job.state -- persist_job_hook_fun's HOOK_DETAIL_SETPREASON/ + * HOOK_DETAIL_CLRPREASON arm captures js.state verbatim from the job's + * CURRENT (unchanged) state, so apply_job_update_transition() reports + * JOB_TRANSITION_NOOP_DUPLICATE on every single real PREASON replay. Gating + * pend_reasons adoption on result==JOB_TRANSITION_APPLIED would therefore + * make it dead code for the realistic, only-ever-occurring case. + * test_wjp_apply_committed_entry_sets_preason_no_state_change below is this + * suite's most important test. + * + * IMPORTANT DIVERGENCE FROM bug-1690/1691's OWN precedent: unlike + * ckptinfo/run_usage, job.pend_reasons has a genuine CLEAR operation + * (do_clr_preason, HOOK_DETAIL_CLRPREASON) that legitimately resets it back + * to NULL and must survive replay just as durably as the SET side. CKPT's/ + * RUSAGE's convention ("payload of exactly sizeof(wal_jobstate_t) == no + * trailing bytes == leave field untouched") cannot distinguish that + * legitimate NULL from an old-format payload, so this fix adds an explicit + * presence byte after the wal_jobstate_t header (0 = cleared, adopt NULL; 1 + * = capnp JobPendReason bytes follow). test_wjp_apply_committed_entry_ + * clears_preason / test_wjp_apply_wal_entry_clears_preason / + * test_wjp_real_clear_survives_restart below are the tests that would have + * silently failed under a naive copy of the CKPT/RUSAGE wire shape. + * + * ALSO DIVERGES FROM bug-1690's WAL_JOB_RUSAGE fix (but matches bug-1691's + * WAL_JOB_CKPT): adoption is a plain free-and-replace (via the capnp- + * generated free_JobPendReason_ptr()), NOT a merge. job.pend_reasons and its + * nested host/resource reason arrays are plain calloc()/strdup()-backed + * throughout -- verified via erase_job_preason()'s FREEUP-only release + * (src/lib/preason.c) and every add_*_preason() constructor (strdup()/ + * calloc()), plus the capnp-generated decode_HostBasedReason/ + * decode_ResourceBasedReason using STRING_DUP+calloc (src/lib/ + * controller.capnp.c) -- NOT xstring/RTTI-backed like ckptinfo/rusage's + * host_name, so no merge-not-replace hazard exists here. + * + * Also proves backward compatibility: a WAL_JOB_PREASON entry with the OLD + * payload shape (exactly sizeof(wal_jobstate_t), no presence byte -- + * simulating an entry written before this fix) still decodes cleanly, with + * pend_reasons left untouched exactly as before this fix -- no crash, no + * misread. + * + * Fixtures mirror test_wal_job_ckpt_fact.c's own conventions exactly: + * - wjp_setup/wjp_teardown: bare cluster_t (jobmgr + hooksys) + in-memory + * persist_t. + * - wjp_make_fresh_persist/wjp_teardown_persist: real disk-backed + * persist_t + persist_recover(), for restart-replay tests. + * + * test_wjp_real_do_set_preason_survives_restart / test_wjp_real_clear_ + * survives_restart drive the REAL, unmodified do_set_preason()/ + * do_clr_preason() (the actual scheduler-reported pend-reason receipt + * handlers) through a real WAL-write + persist_recover() restart cycle -- + * unlike on_job_ckpt()/on_job_finish(), these need a real (if minimal, + * zeroed, no-iomgr) endpoint_t because they call xce_reply() before + * processing, mirroring test_commit_barrier_reply.c's own established + * "zeroed static endpoint_t, fd=-1, packets land on ep->outq" pattern + * (drained here via endpoint_deq()/free_packet() to avoid a reachable-leak + * false positive under the suite's LeakSanitizer pass). set_preason() itself + * is file-static to job.c, so do_set_preason() -- the real production + * entrypoint (job.c's only exported wrapper) -- is what these tests drive, + * exactly as the scheduler's actual OPCode_setpreason/OPCode_clrpreason + * wire path does. + */ +#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 "preason.h" +#include "typemap.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_wjp_jobobj_ready; + +static cluster_t g_wjp_c; +static persist_t g_wjp_p; + +/* ── Spy hook: counts AFTER_EVENT firings, ignores all payload args ─────── */ +static int g_wjp_hook_fires; + +static int wjp_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_wjp_hook_fires++; + return 0; +} + +static void wjp_setup(void) +{ + kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, + .nslots = 32, + .delta = 16 }; + + if (makeString((char*)"wjp-probe") == NULL) + xstring_table_init(64); + + if (!g_wjp_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_wjp_jobobj_ready = 1; + } + + memset(&g_wjp_c, 0, sizeof(g_wjp_c)); + TEST_ASSERT_EQUAL_INT(0, + xce_hook_init(&g_wjp_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); + + g_wjp_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); + TEST_ASSERT_NOT_NULL(g_wjp_c.jobmgr); + + memset(&g_wjp_p, 0, sizeof(g_wjp_p)); + g_wjp_p.cluster = &g_wjp_c; + + g_wjp_hook_fires = 0; +} + +static void wjp_teardown(void) +{ + xce_hook_fini(&g_wjp_c.hooksys); + if (g_wjp_c.jobmgr) + dereference(g_wjp_c.jobmgr); +} + +/* Look up a job_t wrapper by jobid (NULL if absent). */ +static job_t* wjp_find_job(uint64_t jobid) +{ + job_t* j = NULL; + lock(g_wjp_c.jobmgr, RTTI_LOCK_READ); + int found = key_find(g_wjp_c.jobmgr, (void*)jobid, (void**)&j); + unlock(g_wjp_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_ckpt_fact.c's wjc_seed_job. */ +static job_submit_data_t g_wjp_seed_submit; + +static job_t* wjp_seed_job(uint64_t jobid, uint32_t state) +{ + xce_job_t jdef; + + memset(&g_wjp_seed_submit, 0, sizeof(g_wjp_seed_submit)); + g_wjp_seed_submit.queue = (char*)"q"; + g_wjp_seed_submit.user = (char*)"tester"; + + memset(&jdef, 0, sizeof(jdef)); + jdef.jobid = jobid; + jdef.state = state; + jdef.submit = &g_wjp_seed_submit; + + return makeJob(g_wjp_c.jobmgr, &jdef); +} + +/* Build an xce_job_preason_t with one host-based AND one resource-based + * reason (exercises both nested list types, per this bug's audit + * requirement) — caller owns the result (release via + * free_JobPendReason_ptr, mirroring persist.c's own adoption verb, or + * erase_job_preason+FREEUP for a manually-built one). */ +static xce_job_preason_t* wjp_build_preason(int main_reason, + const char* host_name, + int host_reason, + const char* res_name, + const char* res_host, + int res_reason) +{ + xce_job_preason_t* pr = calloc(1, sizeof(*pr)); + TEST_ASSERT_NOT_NULL(pr); + pr->main_reason = main_reason; + + TEST_ASSERT_TRUE( + add_job_preason(pr, host_reason, (char*)host_name, NULL) > 0); + TEST_ASSERT_TRUE( + add_job_preason(pr, res_reason, (char*)res_host, (char*)res_name) + > 0); + + return pr; +} + +static void wjp_assert_preason_matches(job_t* j, int main_reason, + const char* host_name, int host_reason, + const char* res_name, + const char* res_host, int res_reason) +{ + xce_job_preason_t* pr = j->job.pend_reasons; + + TEST_ASSERT_NOT_NULL_MESSAGE(pr, + "a WAL_JOB_PREASON payload with a trailing capnp blob must set " + "job.pend_reasons (bug-1692), not just state"); + TEST_ASSERT_EQUAL_INT(main_reason, pr->main_reason); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, pr->nhost_reasons, + "host-based reason list must survive replay"); + TEST_ASSERT_NOT_NULL(pr->host_based_reasons); + TEST_ASSERT_NOT_NULL(pr->host_based_reasons[0]); + TEST_ASSERT_EQUAL_STRING(host_name, pr->host_based_reasons[0]->host); + TEST_ASSERT_EQUAL_INT(host_reason, pr->host_based_reasons[0]->reason); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, pr->nres_reasons, + "resource-based reason list must survive replay"); + TEST_ASSERT_NOT_NULL(pr->res_based_reasons); + TEST_ASSERT_NOT_NULL(pr->res_based_reasons[0]); + TEST_ASSERT_EQUAL_STRING(res_name, pr->res_based_reasons[0]->resource); + TEST_ASSERT_EQUAL_STRING(res_host, pr->res_based_reasons[0]->host); + TEST_ASSERT_EQUAL_INT(res_reason, pr->res_based_reasons[0]->reason); +} + +/* Local mirror of persist.c's static capnp_encode_jobpreason() -- this + * translation unit cannot reach that file-static function, exactly like + * test_wal_job_ckpt_fact.c's wjc_encode_job mirrors capnp_encode_job. */ +static int wjp_encode_jobpreason(const xce_job_preason_t* pr, uint8_t** out, + size_t* out_len) +{ + struct capn c; + struct capn_segment* cs; + struct JobPendReason cjp; + JobPendReason_ptr jpp; + + capn_init_malloc(&c); + cs = capn_root(&c).seg; + encode_JobPendReason(cs, &cjp, (xce_job_preason_t*)pr); + jpp = new_JobPendReason(cs); + write_JobPendReason(&cjp, jpp); + capn_setp(capn_root(&c), 0, jpp.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; +} + +/* Build a wal_jobstate_t header + presence byte (1) + trailing capnp + * JobPendReason bytes — persist_job_hook_fun's WAL_JOB_PREASON "set" payload + * shape (bug-1692). Caller owns *out. */ +static void wjp_build_set_payload(uint64_t jobid, uint32_t state, + const xce_job_preason_t* pr, + uint8_t** out, size_t* out_len) +{ + wal_jobstate_t js = { .jobid = jobid, .state = state, ._pad = 0 }; + uint8_t* capnp_buf = NULL; + size_t capnp_len = 0; + + TEST_ASSERT_EQUAL_INT(0, + wjp_encode_jobpreason(pr, &capnp_buf, &capnp_len)); + + *out_len = sizeof(js) + 1 + capnp_len; + *out = malloc(*out_len); + TEST_ASSERT_NOT_NULL(*out); + memcpy(*out, &js, sizeof(js)); + (*out)[sizeof(js)] = 1; + if (capnp_len > 0) + memcpy(*out + sizeof(js) + 1, capnp_buf, capnp_len); + free(capnp_buf); +} + +/* Build a wal_jobstate_t header + presence byte (0), no trailing bytes — + * persist_job_hook_fun's WAL_JOB_PREASON "clear" payload shape (bug-1692, + * do_clr_preason's own fact). Caller owns *out. */ +static void wjp_build_clear_payload(uint64_t jobid, uint32_t state, + uint8_t** out, size_t* out_len) +{ + wal_jobstate_t js = { .jobid = jobid, .state = state, ._pad = 0 }; + + *out_len = sizeof(js) + 1; + *out = malloc(*out_len); + TEST_ASSERT_NOT_NULL(*out); + memcpy(*out, &js, sizeof(js)); + (*out)[sizeof(js)] = 0; +} + +/* ── apply_committed_entry(): live-commit apply / boot replay dispatch ──── */ + +static void test_wjp_apply_committed_entry_sets_preason_no_state_change(void) +{ + wjp_setup(); + + job_t* j = wjp_seed_job(75001, JOB_STAT_PEND); + TEST_ASSERT_NOT_NULL(j); + + xce_job_preason_t* pr = + wjp_build_preason(7, "worker-3", 11, "gpu", "worker-4", 12); + + uint8_t* payload = NULL; + size_t plen = 0; + wjp_build_set_payload(75001, JOB_STAT_PEND, pr, &payload, &plen); + erase_job_preason(pr); FREEUP(pr); + + xce_add_hook(&g_wjp_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjp_spy_hook, + NULL); + + int rc = apply_committed_entry(&g_wjp_p, WAL_JOB_PREASON, 400, payload, + (uint32_t)plen); + free(payload); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); + wjp_assert_preason_matches(j, 7, "worker-3", 11, "gpu", "worker-4", 12); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wjp_hook_fires, + "a same-state (duplicate) PREASON replay -- the realistic, " + "only-ever-occurring case, since set_preason() never changes " + "job.state -- must NOT fire a notification hook, unchanged from " + "before this fix, even though pend_reasons WAS adopted"); + + wjp_teardown(); + printf( + "test_wjp_apply_committed_entry_sets_preason_no_state_change: " + "PASS\n"); +} + +/* Secondary scenario: a WAL_JOB_PREASON fact that DOES carry a forward state + * transition (mechanism-level test, mirroring bug-1690/1691's own coverage + * of this same divergence). pend_reasons must be adopted here too, and the + * existing forward-transition hook-fire behavior must be unchanged. */ +static void +test_wjp_apply_committed_entry_sets_preason_forward_transition(void) +{ + wjp_setup(); + + job_t* j = wjp_seed_job(75002, JOB_STAT_PEND); + TEST_ASSERT_NOT_NULL(j); + + xce_job_preason_t* pr = + wjp_build_preason(8, "worker-5", 21, "mem", "worker-6", 22); + + uint8_t* payload = NULL; + size_t plen = 0; + wjp_build_set_payload(75002, JOB_STAT_RUN, pr, &payload, &plen); + erase_job_preason(pr); FREEUP(pr); + + xce_add_hook(&g_wjp_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjp_spy_hook, + NULL); + + int rc = apply_committed_entry(&g_wjp_p, WAL_JOB_PREASON, 401, payload, + (uint32_t)plen); + free(payload); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_RUN, j->job.state); + wjp_assert_preason_matches(j, 8, "worker-5", 21, "mem", "worker-6", 22); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_wjp_hook_fires, + "a forward PREASON-carried transition must still fire exactly one " + "notification, unchanged from before this fix"); + + wjp_teardown(); + printf( + "test_wjp_apply_committed_entry_sets_preason_forward_transition: " + "PASS\n"); +} + +static void +test_wjp_apply_committed_entry_backward_compat_old_payload(void) +{ + wjp_setup(); + + job_t* j = wjp_seed_job(75003, JOB_STAT_PEND); + TEST_ASSERT_NOT_NULL(j); + TEST_ASSERT_NULL_MESSAGE(j->job.pend_reasons, + "fixture must seed a job with pend_reasons still unset"); + + wal_jobstate_t js = { .jobid = 75003, .state = JOB_STAT_PEND, ._pad = 0 }; + + int rc = apply_committed_entry(&g_wjp_p, WAL_JOB_PREASON, 402, &js, + sizeof(js)); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); + TEST_ASSERT_NULL_MESSAGE(j->job.pend_reasons, + "an old-format (exactly sizeof(wal_jobstate_t)) WAL_JOB_PREASON " + "payload must leave pend_reasons NULL -- no crash, no misread of " + "absent trailing bytes"); + + wjp_teardown(); + printf("test_wjp_apply_committed_entry_backward_compat_old_payload: " + "PASS\n"); +} + +/* PREASON-specific (no CKPT/RUSAGE analog): a presence byte of 0 must + * genuinely clear an existing pend_reasons on replay -- proving the + * presence-byte design actually closes the gap a naive copy of bug-1690/ + * 1691's "trailing bytes present or absent" convention would have missed. */ +static void test_wjp_apply_committed_entry_clears_preason(void) +{ + wjp_setup(); + + job_t* j = wjp_seed_job(75004, JOB_STAT_PEND); + TEST_ASSERT_NOT_NULL(j); + j->job.pend_reasons = + wjp_build_preason(9, "stale-host", 1, "stale-res", "stale-host2", 2); + + uint8_t* payload = NULL; + size_t plen = 0; + wjp_build_clear_payload(75004, JOB_STAT_PEND, &payload, &plen); + + int rc = apply_committed_entry(&g_wjp_p, WAL_JOB_PREASON, 403, payload, + (uint32_t)plen); + free(payload); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_NULL_MESSAGE(j->job.pend_reasons, + "a presence-byte=0 (explicit clear) WAL_JOB_PREASON payload must " + "reset pend_reasons to NULL on replay, not merely leave it " + "untouched (bug-1692's key divergence from bug-1690/1691's " + "precedent)"); + + wjp_teardown(); + printf("test_wjp_apply_committed_entry_clears_preason: PASS\n"); +} + +/* present=1 with ZERO trailing bytes (a malformed length no real encoder + * ever produces -- capnp_encode_jobpreason() always emits at least a + * minimal root-pointer+struct-header blob) must NOT be treated as a clear + * (present != 0) and must NOT crash -- decode_jobpreason() is never even + * called (guarded by the length check), so `decoded` stays NULL and the + * prior value must survive untouched, matching every other decode helper's + * failure fallback in apply_committed_entry. (NOTE: unlike a length check, + * capnp-c's read path does not itself validate content -- genuinely + * malformed-but-nonempty bytes decode into a zero-valued struct rather than + * failing, matching decode_joballoc/decode_jobrunusage's identical + * leniency, so that scenario is not distinguishable from a real empty + * JobPendReason at this layer; WAL entry CRC framing is what protects + * against on-disk corruption, not this logical decode step.) */ +static void +test_wjp_apply_committed_entry_decode_failure_leaves_untouched(void) +{ + wjp_setup(); + + job_t* j = wjp_seed_job(75005, JOB_STAT_PEND); + TEST_ASSERT_NOT_NULL(j); + j->job.pend_reasons = + wjp_build_preason(4, "keep-host", 5, "keep-res", "keep-host2", 6); + + wal_jobstate_t js = { .jobid = 75005, .state = JOB_STAT_PEND, ._pad = 0 }; + uint8_t payload[sizeof(js) + 1]; + memcpy(payload, &js, sizeof(js)); + payload[sizeof(js)] = 1; /* present, but no capnp bytes follow at all */ + + int rc = apply_committed_entry(&g_wjp_p, WAL_JOB_PREASON, 404, payload, + (uint32_t)sizeof(payload)); + + TEST_ASSERT_EQUAL_INT(0, rc); + wjp_assert_preason_matches(j, 4, "keep-host", 5, "keep-res", + "keep-host2", 6); + + wjp_teardown(); + printf( + "test_wjp_apply_committed_entry_decode_failure_leaves_untouched: " + "PASS\n"); +} + +/* ── apply_wal_entry(): standby catch-up / replicated-apply dispatch ────── */ + +static void test_wjp_apply_wal_entry_sets_preason_no_hook(void) +{ + wjp_setup(); + + job_t* j = wjp_seed_job(75006, JOB_STAT_PEND); + TEST_ASSERT_NOT_NULL(j); + + xce_job_preason_t* pr = + wjp_build_preason(3, "worker-7", 31, "disk", "worker-8", 32); + + uint8_t* payload = NULL; + size_t plen = 0; + wjp_build_set_payload(75006, JOB_STAT_PEND, pr, &payload, &plen); + erase_job_preason(pr); FREEUP(pr); + + xce_add_hook(&g_wjp_c.hooksys, XCE_CONTROLLER_HOOK_JOB, wjp_spy_hook, + NULL); + + int rc = apply_wal_entry(&g_wjp_p, WAL_JOB_PREASON, 405, payload, + (uint32_t)plen); + free(payload); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); + wjp_assert_preason_matches(j, 3, "worker-7", 31, "disk", "worker-8", 32); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wjp_hook_fires, + "apply_wal_entry's catch-up dispatch must never fire a " + "notification hook"); + + wjp_teardown(); + printf("test_wjp_apply_wal_entry_sets_preason_no_hook: PASS\n"); +} + +static void test_wjp_apply_wal_entry_backward_compat_old_payload(void) +{ + wjp_setup(); + + job_t* j = wjp_seed_job(75007, JOB_STAT_PEND); + TEST_ASSERT_NOT_NULL(j); + TEST_ASSERT_NULL_MESSAGE(j->job.pend_reasons, + "fixture must seed a job with pend_reasons still unset"); + + wal_jobstate_t js = { .jobid = 75007, .state = JOB_STAT_PEND, ._pad = 0 }; + + int rc = apply_wal_entry(&g_wjp_p, WAL_JOB_PREASON, 406, &js, sizeof(js)); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); + TEST_ASSERT_NULL_MESSAGE(j->job.pend_reasons, + "an old-format WAL_JOB_PREASON payload must leave pend_reasons " + "NULL on this seam too"); + + wjp_teardown(); + printf("test_wjp_apply_wal_entry_backward_compat_old_payload: PASS\n"); +} + +static void test_wjp_apply_wal_entry_clears_preason(void) +{ + wjp_setup(); + + job_t* j = wjp_seed_job(75008, JOB_STAT_PEND); + TEST_ASSERT_NOT_NULL(j); + j->job.pend_reasons = + wjp_build_preason(2, "stale-a", 1, "stale-b", "stale-c", 2); + + uint8_t* payload = NULL; + size_t plen = 0; + wjp_build_clear_payload(75008, JOB_STAT_PEND, &payload, &plen); + + int rc = apply_wal_entry(&g_wjp_p, WAL_JOB_PREASON, 407, payload, + (uint32_t)plen); + free(payload); + + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_NULL_MESSAGE(j->job.pend_reasons, + "a presence-byte=0 WAL_JOB_PREASON payload must clear " + "pend_reasons on the standby catch-up seam too"); + + wjp_teardown(); + printf("test_wjp_apply_wal_entry_clears_preason: PASS\n"); +} + +/* ── Restart-replay observables ──────────────────────────────────────── */ + +static persist_t* wjp_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 wjp_teardown_persist(persist_t* p) +{ + persist_stop(p); + persist_fini(p); + free(p); +} + +/* Encode helper mirroring test_wal_job_ckpt_fact.c's wjc_encode_job + * (capnp_encode_job is file-static in persist.c). */ +static int wjp_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_PREASON (real host+resource reasons) straight onto disk via + * persist_wal_async, then replays via a fresh persist_t's persist_recover() + * (a real restart, exercising apply_committed_entry's boot-replay dispatch). + * Isolates the WAL fact shape from the live caller, mirroring + * test_wal_job_ckpt_fact.c's own restart-replay test. */ +static void test_wjp_restart_replay_recovers_preason(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_wjp_replay_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + /* Phase 1: write the log to real disk. */ + { + persist_t* p = wjp_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 = 75009; + jdef.state = JOB_STAT_PEND; + jdef.submit = &submit; + + uint8_t* job_payload = NULL; + size_t job_plen = 0; + TEST_ASSERT_EQUAL_INT(0, + wjp_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_preason_t* pr = + wjp_build_preason(6, "replay-host", 41, "replay-res", + "replay-host2", 42); + uint8_t* pr_payload = NULL; + size_t pr_plen = 0; + wjp_build_set_payload(75009, JOB_STAT_PEND, pr, &pr_payload, + &pr_plen); + erase_job_preason(pr); FREEUP(pr); + + TEST_ASSERT_EQUAL_INT(0, + persist_wal_async(p, WAL_JOB_PREASON, pr_payload, + (uint32_t)pr_plen)); + free(pr_payload); + + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + wjp_teardown_persist(p); + } + + /* Phase 2: fresh persist_t + fresh cluster_t, recover from disk — + * simulates a controller restart. */ + { + persist_t* p2 = wjp_make_fresh_persist(tmp); + TEST_ASSERT_NOT_NULL(p2); + + wjp_setup(); + p2->cluster = &g_wjp_c; + + int rc = persist_recover(p2); + TEST_ASSERT_EQUAL_INT(0, rc); + + job_t* j = wjp_find_job(75009); + TEST_ASSERT_NOT_NULL_MESSAGE(j, + "restart replay must construct the job via WAL_JOB_SUBMIT"); + TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j->job.state); + wjp_assert_preason_matches(j, 6, "replay-host", 41, "replay-res", + "replay-host2", 42); + + wjp_teardown(); + wjp_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_wjp_restart_replay_recovers_preason: PASS\n"); +} + +/* Drain an endpoint's reply queue (avoids a reachable-leak false positive + * under this suite's LeakSanitizer pass — mirrors test_commit_barrier_ + * reply.c's cbrp_scan_replies, minus the reply-content decode this suite + * doesn't need). */ +static void wjp_drain_ep(endpoint_t* ep) +{ + data_item_t* it; + while ((it = endpoint_deq(ep)) != NULL) { + io_packet_t* pkt = ITPTR_TO_PACKET(it); + free_packet(pkt); + free(pkt); + } +} + +/* bug-1692's primary regression test: drives the REAL production + * do_set_preason() path (the actual scheduler-reported pend-reason receipt + * handler — job.c's exported entrypoint for OPCode_setpreason; set_preason() + * itself is file-static) with a real xce_job_preason_t, through a real + * WAL-write + persist_recover() restart cycle — the audit's exact confirmed + * root cause. do_set_preason() calls xce_reply() before processing, so a + * minimal zeroed (fd=-1, no iomgr) endpoint_t is needed, mirroring + * test_commit_barrier_reply.c's established fixture pattern. */ +static void test_wjp_real_do_set_preason_survives_restart(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_wjp_onset_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + const uint64_t jobid = 75010; + + /* 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 do_set_preason() call site has something to + * find/lock) -- mirrors test_wjc_real_on_job_ckpt_survives_restart's + * dual-seed pattern exactly. Then drive an actual do_set_preason() call + * and let IT append WAL_JOB_PREASON 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*)"wjp-onset-probe") == NULL) + xstring_table_init(64); + if (!g_wjp_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_wjp_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); + c.myhost = &c.myhost_info; + c.myhost_info.name = (char*)"ctrl-test"; + c.myhost_info.key = (char*)""; + + cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + persist_set_workdir(&c.persist, tmp, "ctrl-test"); + /* Same reasoning as test_wjc_real_on_job_ckpt_survives_restart: + * do_set_preason()/set_preason() (job.c) are hook-driven and rely + * ENTIRELY on persist_job_hook_fun being registered as a c.hooksys + * subscriber to reach the WAL at all -- persist_start() only + * registers that hook `if (p->cluster != NULL)`. */ + 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_PEND; + wal_jdef.submit = &wal_submit; + + TEST_ASSERT_EQUAL_INT(0, + wjp_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 for do_set_preason()/set_preason() 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_PEND; + jdef.submit = &submit; + + TEST_ASSERT_NOT_NULL(makeJob(c.jobmgr, &jdef)); + } + + /* The real scheduler-reported pend-reason set. jpr is heap- + * allocated and MUST be, since set_preason() MOVE_PTRs it straight + * into the job (job.c) -- mirrors test_commit_barrier_reply.c's + * cbrp_arrange_set_preason's own doc comment on this exact point. */ + xce_job_preason_t* jpr = + wjp_build_preason(5, "real-host", 51, "real-res", "real-host2", + 52); + + xce_pendreason_t pend = { .jobid = jobid, .reasons = jpr }; + xce_pendreason_t* prs[1] = { &pend }; + xce_preason_set_t set; + memset(&set, 0, sizeof(set)); + set.nreasons = 1; + set.reasons = prs; + + xce_peerinfo_t from = { .name = (char*)"scheduler", .key = (char*)"" }; + xce_request_t req; + memset(&req, 0, sizeof(req)); + req.opcode = OPCode_setpreason; + req.seq = 1; + req.from = &from; + req.body_which = Request_body_preasonmsg; + req.body.preasonmsg = &set; + + endpoint_t ep; + memset(&ep, 0, sizeof(ep)); + ep.fd = -1; + + int rc = do_set_preason(&c, &ep, &req); + TEST_ASSERT_EQUAL_INT(0, rc); + wjp_drain_ep(&ep); + + 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 = wjp_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_PEND, j2->job.state, + "restart replay must recover the job's state, unchanged by the " + "real do_set_preason() call"); + + /* bug-1692: restart replay must recover the scheduler-reported + * pend_reasons -- the audit's exact confirmed root cause (mirrors + * bug-1678/1687/1689/1690/1691's own acceptance criterion). */ + wjp_assert_preason_matches(j2, 5, "real-host", 51, "real-res", + "real-host2", 52); + + xce_hook_fini(&c2.hooksys); + if (c2.jobmgr) dereference(c2.jobmgr); + wjp_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_wjp_real_do_set_preason_survives_restart: PASS\n"); +} + +/* PREASON-specific (no CKPT/RUSAGE analog): drives the REAL do_set_preason() + * THEN the REAL do_clr_preason() before the restart, proving the clear + * genuinely survives replay too -- not just the set. This is the scenario a + * naive copy of bug-1690/1691's "trailing bytes present or absent" wire + * convention would have silently gotten wrong (a stale, pre-clear + * pend_reasons would resurrect on restart). */ +static void test_wjp_real_clear_survives_restart(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_wjp_onclr_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + const uint64_t jobid = 75011; + + { + persist_config_t cfg; + + if (makeString((char*)"wjp-onclr-probe") == NULL) + xstring_table_init(64); + if (!g_wjp_jobobj_ready) { + TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); + g_wjp_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); + c.myhost = &c.myhost_info; + c.myhost_info.name = (char*)"ctrl-test"; + c.myhost_info.key = (char*)""; + + 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, &c)); + TEST_ASSERT_EQUAL_INT(0, persist_start(&c.persist)); + + { + 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_PEND; + wal_jdef.submit = &wal_submit; + + TEST_ASSERT_EQUAL_INT(0, + wjp_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); + } + + { + 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_PEND; + jdef.submit = &submit; + + TEST_ASSERT_NOT_NULL(makeJob(c.jobmgr, &jdef)); + } + + endpoint_t ep; + memset(&ep, 0, sizeof(ep)); + ep.fd = -1; + + /* SET first (real path), then CLEAR (real path) -- both must be + * durably reflected by the time we restart below. */ + { + xce_job_preason_t* jpr = + wjp_build_preason(1, "temp-host", 61, "temp-res", + "temp-host2", 62); + xce_pendreason_t pend = { .jobid = jobid, .reasons = jpr }; + xce_pendreason_t* prs[1] = { &pend }; + xce_preason_set_t set; + memset(&set, 0, sizeof(set)); + set.nreasons = 1; + set.reasons = prs; + + xce_peerinfo_t from = { .name = (char*)"scheduler", + .key = (char*)"" }; + xce_request_t req; + memset(&req, 0, sizeof(req)); + req.opcode = OPCode_setpreason; + req.seq = 1; + req.from = &from; + req.body_which = Request_body_preasonmsg; + req.body.preasonmsg = &set; + + TEST_ASSERT_EQUAL_INT(0, do_set_preason(&c, &ep, &req)); + wjp_drain_ep(&ep); + } + + { + uint64_t jobs[1] = { jobid }; + xce_clear_reason_t clr = { .njobs = 1, .jobs = jobs }; + xce_peerinfo_t from = { .name = (char*)"scheduler", + .key = (char*)"" }; + xce_request_t req; + memset(&req, 0, sizeof(req)); + req.opcode = OPCode_clrpreason; + req.seq = 2; + req.from = &from; + req.body_which = Request_body_clrpreasonmsg; + req.body.clrpreasonmsg = &clr; + + TEST_ASSERT_EQUAL_INT(0, do_clr_preason(&c, &ep, &req)); + wjp_drain_ep(&ep); + } + + 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); + } + + { + persist_t* p2 = wjp_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"); + + /* bug-1692's key correctness fix: the CLEAR (do_clr_preason) must + * durably survive replay, not just the earlier SET -- a stale + * pend_reasons resurrecting here would be exactly the bug a naive + * copy of bug-1690/1691's wire convention would have produced. */ + TEST_ASSERT_NULL_MESSAGE(j2->job.pend_reasons, + "restart replay must recover the CLEARED pend_reasons state " + "(NULL), not resurrect the earlier SET's stale value"); + + xce_hook_fini(&c2.hooksys); + if (c2.jobmgr) dereference(c2.jobmgr); + wjp_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_wjp_real_clear_survives_restart: PASS\n"); +} + +/* ── Test group entry point ───────────────────────────────────────────── */ + +void test_wal_job_preason_fact(void) +{ + UnitySetTestFile(__FILE__); + RUN_TEST(test_wjp_apply_committed_entry_sets_preason_no_state_change); + RUN_TEST(test_wjp_apply_committed_entry_sets_preason_forward_transition); + RUN_TEST(test_wjp_apply_committed_entry_backward_compat_old_payload); + RUN_TEST(test_wjp_apply_committed_entry_clears_preason); + RUN_TEST(test_wjp_apply_committed_entry_decode_failure_leaves_untouched); + RUN_TEST(test_wjp_apply_wal_entry_sets_preason_no_hook); + RUN_TEST(test_wjp_apply_wal_entry_backward_compat_old_payload); + RUN_TEST(test_wjp_apply_wal_entry_clears_preason); + RUN_TEST(test_wjp_restart_replay_recovers_preason); + RUN_TEST(test_wjp_real_do_set_preason_survives_restart); + RUN_TEST(test_wjp_real_clear_survives_restart); +} diff --git a/src/include/wal_types.h b/src/include/wal_types.h index dc1a874f..365ffae1 100644 --- a/src/include/wal_types.h +++ b/src/include/wal_types.h @@ -187,7 +187,43 @@ typedef enum { * 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_PREASON = 10, /* scheduler-reported host/resource-based pending + * reasons (job.pend_reasons, set by set_preason()/ + * cleared by do_clr_preason(), job.c). Payload is + * wal_jobstate_t{jobid, state, _pad=0} for any + * entry written before bug-1692. As of bug-1692, + * persist_job_hook_fun's HOOK_DETAIL_SETPREASON/ + * HOOK_DETAIL_CLRPREASON arm appends an explicit + * presence byte after the header: 0 = pend_reasons + * was genuinely cleared (adopt NULL on replay), 1 + * = trailing Cap'n Proto JobPendReason bytes + * follow (adopt the decoded value). Unlike + * WAL_JOB_CKPT/WAL_JOB_RUSAGE, "trailing bytes + * present or absent" alone cannot distinguish + * "old-format payload" from "field is legitimately + * NULL", because pend_reasons -- uniquely among + * this fact-type family -- has a real CLEAR + * operation (do_clr_preason) that must survive + * replay durably, not just its SET side; the + * presence byte resolves that ambiguity. A payload + * of exactly sizeof(wal_jobstate_t) (no presence + * byte at all) leaves pend_reasons untouched on + * both apply seams, matching the pre-fix + * state-only behavior exactly. Like + * WAL_JOB_RUSAGE/WAL_JOB_CKPT, adoption on the + * apply_committed_entry() seam is NOT gated on a + * forward state transition -- set_preason()/ + * do_clr_preason() never change job.state, so + * gating it the same way would make it dead code. + * Adoption is a plain free-and-replace (via the + * capnp-generated free_JobPendReason_ptr()), NOT a + * merge like WAL_JOB_RUSAGE -- job.pend_reasons + * and its nested host/resource reason arrays are + * plain calloc()/strdup() throughout (verified via + * erase_job_preason()'s FREEUP-only release, NOT + * xstring/RTTI-backed like ckptinfo/rusage's + * host_name), so no merge-not-replace hazard + * exists here. */ WAL_JOB_ARRAY_SUBMIT = 11, /* job-array-single-wal-record: one record for * the whole array (base spec + index range), * distinct from WAL_JOB_SUBMIT's single-job