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

fix(job-array-single-wal-record): extend Live Apply Resource Fence to boot replay (task 4.2's gap)

Browse source 
replay_wal() now halts logical apply and engages persist_mark_apply_resource_fenced()
on a negative apply_committed_entry() return during boot/restart replay, mirroring
persist_apply_pump_run()'s existing live-path halt-and-fence behavior (task 4.1).
Physical scan bookkeeping (current_lsn, CRC chain) stays unconditional per entry,
consistent with persist_recover()'s own established current_lsn/last_applied_lsn
divergence precedent.

Went through one remediation round: round 1 exercised the wrong ha.c gate (periodic
re-alert timer); round 2 corrected the test to drive the real election-time gate
(raft_cb_become_leader_with_cluster_for_test()), with teeth independently confirmed
via a temporary fence-clear RED/GREEN cycle. Round-2 reviewer APPROVED.

Documents a real, NOT-fixed residual gap found during this task: a node's very
first-ever promotion evaluation in a fresh process runs before that node's own
first persist_recover()/replay_wal() call, so this fence is invisible to that one
promotion check (ha.c:1303-1408 gate runs before ha.c:1427-1439's first-boot
recovery). Every promotion evaluation after a process's first one is protected.
Out of this task's persist.c-only boundary; flagged for a dedicated ha.c follow-up.
This commit is contained in:
Rongsong Shen 2026-07-16 16:39:07 +00:00
parent 5a8136263e
commit ea8d9d3c4f
5 changed files with 669 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

1
src/controller/meson.build
View file

@ -167,6 +167,7 @@ controller_unit_test_src = [
'testsuite/test_apply_pump_integration.c',
'testsuite/test_job_array_apply.c',
'testsuite/test_job_array_fence.c',
'testsuite/test_job_array_replay_fence.c',
'ctrlstatus.c',
'commit.c',
'job.c',

229
src/controller/persist.c
View file

@ -3731,6 +3731,21 @@ static int replay_wal(persist_t* p, uint64_t from_lsn,
int replayed = 0;
uint32_t last_crc = WAL_CRC_NONE;
/* job-array-single-wal-record task 7.2 (design.md "Live Apply Resource
* Fence", task 4.2's self-flagged gap): once a negative
* apply_committed_entry() return is hit during this replay pass, every
* SUBSEQUENT entry is skipped for LOGICAL apply purposes -- mirroring
* persist_apply_pump_run()'s halt-and-fence branch, which never
* dispatches past a failed LSN either. This preserves the strict
* sequential-no-skip apply invariant (task 4.1): a later entry may be
* causally dependent on the one that never actually applied (e.g. a
* job-state fact for an array element that was never constructed).
* The PHYSICAL scan (CRC chain, current_lsn, seg_off) below is
* deliberately NOT gated on this flag -- see the apply_rc < 0 branch's
* doc comment for why current_lsn/last_crc must stay accurate for the
* entire on-disk log regardless of the logical apply outcome. */
bool apply_halted = false;
for (int si = 0; si < nseg; si++) {
char seg_path[4096];
snprintf(seg_path, sizeof(seg_path),
@ -3900,66 +3915,176 @@ static int replay_wal(persist_t* p, uint64_t from_lsn,
* connected endpoint that is empty at first-boot replay time.
* MUST be re-verified before tasks 4.1/6.1/6.2/6.3 make this
* path broadly live. */
int apply_rc = apply_committed_entry(p, hdr.type, hdr.lsn, payload,
hdr.payload_len);
int apply_rc = 0;
if (!apply_halted) {
apply_rc = apply_committed_entry(p, hdr.type, hdr.lsn,
payload, hdr.payload_len);
}
free(payload);
/* raft-apply-decoupling task 5.1: track the replay-derived
* LOGICAL apply cursor separately from current_lsn (the physical
* WAL position owned by this loop, bug-778) — last_applied_lsn is
* the "highest entry actually dispatched via
* apply_committed_entry()" position the post-replay verification
* compares against. Segments are scanned in ascending LSN order
* (sorted above) and entries within a segment are sequential, so
* a plain store (no monotonic max needed) is correct here. */
atomic_store_explicit(&p->last_applied_lsn, hdr.lsn,
memory_order_relaxed);
if (apply_halted) {
/* job-array-single-wal-record task 7.2: an earlier entry in
* THIS replay pass already fenced (see the apply_rc < 0
* branch below) -- this entry and every later one is
* skipped for logical apply, exactly like the
* WAL_CHECKPOINT/unknown-type skip branches above. Physical
* bookkeeping (current_lsn/seg_off) still runs
* unconditionally below. */
} else if (apply_rc < 0) {
/* job-array-single-wal-record task 7.2 (design.md "Live
* Apply Resource Fence", task 4.2's self-flagged residual
* gap): mirror persist_apply_pump_run()'s halt-and-fence
* branch exactly. This is never a content-derived failure
* (Req 9.1-9.3's pre-commit validation already made those
* impossible post-commit) -- it is the SAME node-local
* resource-exhaustion class (Req 9.4) the live pump already
* guards against, just hit during boot/restart replay
* instead of live apply. Do NOT advance last_applied_lsn
* past this LSN (skip that store entirely -- last_applied_lsn
* stays parked at this entry's predecessor) and do NOT tally
* this entry's job count; engage the SAME
* persist_mark_apply_resource_fenced() primitive the live
* path uses -- reused verbatim, no parallel mechanism. This
* closes task 4.2's documented residual gap: without this, a
* node that crashes while fenced and re-hits the identical
* resource-exhaustion condition on restart would silently
* discard the failure here (as this branch previously did,
* unconditionally advancing last_applied_lsn regardless of
* apply_rc's sign) and boot up looking fully promotable
* while actually divergent/behind.
*
* Engaging synchronously HERE, inside this boot-time call
* (persist_recover(), called synchronously from
* cluster_init_phase2() before it returns), matters:
* persist_apply_pump_kick()'s post-recovery catch-up kick
* dispatches ASYNCHRONOUSLY onto a pool worker vtask (or
* only runs inline as its own documented fallback) -- relying
* on that kick alone to eventually re-fail and set the fence
* would leave a real window, between persist_recover()
* returning and that async retry actually running, where a
* promotion-gate check could observe "not fenced" and
* wrongly treat this node as eligible.
*
* KNOWN BOUNDARY (task 7.2 adversarial review, NOT fixed by
* this task -- ha.c change required, out of this function's
* scope): this fence, once engaged here, protects every
* SUBSEQUENT promotion evaluation in this process's life --
* a later election's raft_cb_become_leader() gate (ha.c:1337
* reads persist_apply_resource_fenced(), inside the
* eligibility block at ha.c:1303-1408) and the periodic
* re-alert timer (promotion_fence_timeout_fn, ha.c ~876)
* both correctly observe it. It does NOT protect this SAME
* boot's own first-ever promotion evaluation: on a node's
* first-ever election evaluation in its process's life (i.e.
* BEFORE cluster_init_phase2() has ever run), ha.c's
* raft_cb_become_leader() runs its eligibility gate at
* ha.c:1303-1408 strictly BEFORE the is_first_boot block at
* ha.c:1427-1439 that calls cluster_init_phase2() -- which is
* what calls persist_recover()/replay_wal() (cluster.c:1482)
* and is what would engage this fence for the first time.
* There is no re-check of persist_apply_resource_fenced()
* after cluster_init_phase2() returns; the code proceeds
* straight to CTRL_PRIMARY. So a fence engaged by THIS SAME
* first replay is invisible to the one promotion check
* gating that specific boot -- exactly the "all nodes
* restart simultaneously" race this whole task exists to
* guard against. Closing this requires an ha.c change (a
* re-check of persist_apply_resource_fenced() after
* cluster_init_phase2() returns, or reordering recovery
* ahead of the promotion decision), which is out of this
* spec's current scope -- see tasks.md's task 7.2
* Implementation Notes for the full finding. */
ERROR("replay_wal: apply_committed_entry failed for "
"committed entry lsn=%" PRIu64 " wal_type=%" PRIu32
" (rc=%d) during boot/restart replay -- halting this "
"replay's LOGICAL apply at this position (the entry "
"stays committed and will be retried by the live apply "
"pump's post-recovery kick, from the same position); "
"Live Apply Resource Fence ENGAGED "
"(job-array-single-wal-record Req 9.4/9.5)",
hdr.lsn, hdr.type, apply_rc);
persist_mark_apply_resource_fenced(p, hdr.lsn);
apply_halted = true;
} else {
/* raft-apply-decoupling task 5.1: track the replay-derived
* LOGICAL apply cursor separately from current_lsn (the
* physical WAL position owned by this loop, bug-778) --
* last_applied_lsn is the "highest entry actually dispatched
* via apply_committed_entry()" position the post-replay
* verification compares against. Segments are scanned in
* ascending LSN order (sorted above) and entries within a
* segment are sequential, so a plain store (no monotonic max
* needed) is correct here. */
atomic_store_explicit(&p->last_applied_lsn, hdr.lsn,
memory_order_relaxed);
/* Log-derived half of the post-replay apply-correctness
* expectation (Req 8.1): every WAL_JOB_SUBMIT that reaches this
* point is, by construction, above the loaded snapshot's
* checkpoint_lsn (the from_lsn skip above already filtered out
* anything the snapshot covers) — so it names a job that must now
* exist in the live job table that did not exist before this
* replay pass began. A duplicate WAL_JOB_SUBMIT for an
* already-existing jobid (bug-1419: timeout-then-late-commit
* retry) is a legitimate no-op — apply_committed_entry() returns
* 0 for it, not 1 — so it must NOT be tallied here or the
* post-replay comparison against jobmgr's real growth spuriously
* fails (APPLY_DIVERGENT). */
if (hdr.type == WAL_JOB_SUBMIT && new_job_submits_out &&
apply_rc == 1)
(*new_job_submits_out)++;
/* Log-derived half of the post-replay apply-correctness
* expectation (Req 8.1): every WAL_JOB_SUBMIT that reaches
* this point is, by construction, above the loaded
* snapshot's checkpoint_lsn (the from_lsn skip above already
* filtered out anything the snapshot covers) — so it names a
* job that must now exist in the live job table that did not
* exist before this replay pass began. A duplicate
* WAL_JOB_SUBMIT for an already-existing jobid (bug-1419:
* timeout-then-late-commit retry) is a legitimate no-op —
* apply_committed_entry() returns 0 for it, not 1 — so it
* must NOT be tallied here or the post-replay comparison
* against jobmgr's real growth spuriously fails
* (APPLY_DIVERGENT). */
if (hdr.type == WAL_JOB_SUBMIT && new_job_submits_out &&
apply_rc == 1)
(*new_job_submits_out)++;
/* job-array-single-wal-record task 6.0a: the array analog of the
* WAL_JOB_SUBMIT tally immediately above. apply_committed_entry()'s
* WAL_JOB_ARRAY_SUBMIT case (restore_job_array()'s return,
* generalized to N elements) returns apply_rc == the genuine
* count of NEWLY created elements (1..array_size) on a real
* apply, or 0 for an idempotent whole-record re-apply (every
* index already existed -- restore_job_array() returned
* -EEXIST, which this same dispatch maps to 0, mirroring
* WAL_JOB_SUBMIT's own dedup return). Tallying apply_rc (not a
* flat +1) is required here, unlike the single-job case above,
* because one WAL_JOB_ARRAY_SUBMIT entry can materialize many
* jobmgr entries at once; guarding on apply_rc > 0 excludes the
* idempotent 0 case from double-counting a re-apply, exactly as
* apply_rc == 1 excludes WAL_JOB_SUBMIT's duplicate-jobid case
* (bug-1419) above. Without this branch, replay_wal() silently
* undercounts every array record's real N-element jobmgr growth
* to 0, and persist_verify_apply_correctness() then reports a
* false APPLY_DIVERGENT verdict that engages the PERMANENT,
* non-self-healing persist_recover_loud_fence_if_unrecoverable()
* fence (live-reproduced task 6.1 soak, 2026-07-15). */
if (hdr.type == WAL_JOB_ARRAY_SUBMIT && new_job_submits_out &&
apply_rc > 0)
(*new_job_submits_out) += (uint64_t)apply_rc;
/* job-array-single-wal-record task 6.0a: the array analog of
* the WAL_JOB_SUBMIT tally immediately above.
* apply_committed_entry()'s WAL_JOB_ARRAY_SUBMIT case
* (restore_job_array()'s return, generalized to N elements)
* returns apply_rc == the genuine count of NEWLY created
* elements (1..array_size) on a real apply, or 0 for an
* idempotent whole-record re-apply (every index already
* existed -- restore_job_array() returned -EEXIST, which
* this same dispatch maps to 0, mirroring WAL_JOB_SUBMIT's
* own dedup return). Tallying apply_rc (not a flat +1) is
* required here, unlike the single-job case above, because
* one WAL_JOB_ARRAY_SUBMIT entry can materialize many jobmgr
* entries at once; guarding on apply_rc > 0 excludes the
* idempotent 0 case from double-counting a re-apply, exactly
* as apply_rc == 1 excludes WAL_JOB_SUBMIT's duplicate-jobid
* case (bug-1419) above. Without this branch, replay_wal()
* silently undercounts every array record's real N-element
* jobmgr growth to 0, and persist_verify_apply_correctness()
* then reports a false APPLY_DIVERGENT verdict that engages
* the PERMANENT, non-self-healing
* persist_recover_loud_fence_if_unrecoverable() fence
* (live-reproduced task 6.1 soak, 2026-07-15). */
if (hdr.type == WAL_JOB_ARRAY_SUBMIT && new_job_submits_out &&
apply_rc > 0)
(*new_job_submits_out) += (uint64_t)apply_rc;
/* Advance current_lsn to one past this entry */
replayed++;
}
/* Advance current_lsn to one past this entry -- unconditional,
* regardless of the LOGICAL apply outcome above (mirrors the
* WAL_CHECKPOINT/unknown-type skip branches earlier in this
* loop, which also advance current_lsn without ever dispatching
* through apply_committed_entry()). job-array-single-wal-record
* task 7.2: the physical WAL scan must account for every
* on-disk entry so current_lsn/last_crc stay correct for any
* future append this process makes (as a promoted primary, or
* as a replication standby receiving new entries) -- halting
* this advance too (treating a fenced entry as physically
* absent) would understate the true on-disk tail and risk a
* later append colliding with/overwriting already-durable
* entries beyond the fenced position (persist_recover()'s own
* "Round-2 fix" comment above, for the durable_lsn/apply_cursor
* split, already establishes that current_lsn and
* last_applied_lsn are expected to diverge -- this is the same
* principle, applied to a halted apply instead of a skipped
* checkpoint marker). */
if (hdr.lsn + 1 > p->current_lsn)
p->current_lsn = hdr.lsn + 1;
replayed++;
seg_off += (uint32_t)(sizeof(hdr) + hdr.payload_len);
}

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

@ -87,6 +87,7 @@ extern void test_apply_pump(void);
extern void test_apply_pump_integration(void);
extern void test_job_array_apply(void);
extern void test_job_array_fence(void);
extern void test_job_array_replay_fence(void);
void setUp(void) { return; }
@ -243,6 +244,7 @@ int main(int argc, char* argv[])
TEST(test_apply_pump_integration);
TEST(test_job_array_apply);
TEST(test_job_array_fence);
TEST(test_job_array_replay_fence);
return UNITY_END();
}

485
src/controller/testsuite/test_job_array_replay_fence.c Normal file
View file

@ -0,0 +1,485 @@
/* test_job_array_replay_fence.c — job-array-single-wal-record task 7.2
*
* Closes task 4.2's self-flagged residual gap: persist_apply_pump_run() (the
* LIVE apply path, task 4.1/4.2) halts-and-fences on a negative
* apply_committed_entry() return, but replay_wal() (the BOOT/RESTART REPLAY
* path) previously only captured that return value for the job-count tally
* (task 6.0a) — it never halted or engaged the Live Apply Resource Fence.
* A node that crashed while fenced and restarted into a boot replay hitting
* the IDENTICAL resource-exhaustion condition would silently pass it and
* look fully promotable while actually divergent/behind.
*
* ── Scenario ("crash while fenced, restart into the same condition") ──────
* Phase 1 ("boot 1"): a bare persist_t (no cluster) appends ONE real,
* physical WAL_JOB_ARRAY_SUBMIT entry via persist_wal_async, then tears
* down WITHOUT ever applying it — simulating a crash before the live apply
* pump got a chance to run (e.g. the process died right after append,
* before any apply kick).
*
* Phase 2 ("boot 2" / restart): a fresh cluster_t + its embedded persist_t
* are pointed at the SAME on-disk workdir. The construction-phase
* fault-injection seam (job_array_apply_test_set_force_fail_at — the SAME
* seam task 4.4's capstone test uses, chosen for the identical reason: it
* is the one seam Req 9.6's "no partial/divergent set of elements" claim
* is provably true for, unlike the bulk-insert-phase seam) is armed BEFORE
* persist_recover() runs, so replay_wal() hits the identical
* resource-exhaustion condition during REPLAY, not live apply.
*
* ── Seam choice ─────────────────────────────────────────────────────────
* See test_job_array_fence.c's header comment for the full rationale —
* this test reuses the exact same construction-phase seam for the exact
* same reason (the bulk-insert-phase seam's documented residual atomicity
* gap would make a "no partial set" assertion dishonest here too).
*
* ── Fixture ─────────────────────────────────────────────────────────────
* Mirrors test_replay_apply_seam.c's persist_recover()-driven fixture
* (raft-apply-decoupling task 2.3) for phase 1/2, and test_job_array_fence.c's
* tf_setup ha_init+delaymgr shape (task 4.4) for the promotion-gate
* confirmation — reusing established patterns rather than inventing a new
* fixture style. Deliberately OMITS the commit engine / repl standbys
* test_job_array_fence.c wires: this test drives the physical WAL directly
* via persist_wal_async (like test_replay_apply_seam.c), never through the
* commit engine, so those collaborators are unnecessary here.
*
* IMPORTANT fixture detail: the persist_t used for phase 2 MUST be the one
* EMBEDDED in the cluster_t (&g_trf_c.persist), not a standalone heap
* persist_t — both raft_cb_become_leader_with_cluster_for_test(&g_trf_c, ...)
* and ha_arm_promotion_fence_timer_for_test(&g_trf_c, ...) (and the
* promotion_fence_timeout_fn the latter arms) read the fence state off
* c->persist directly, so the replay under test and the promotion-gate
* checks must observe the exact same persist_t instance.
*
* ── Remediation round 1 (task 7.2 adversarial review) ───────────────────
* The refusal proof below now drives raft_cb_become_leader_with_cluster_
* for_test() — the REAL election-time eligibility gate (ha.c:1303-1408) —
* instead of ha_arm_promotion_fence_timer_for_test() (the periodic re-alert
* timer, ha.c ~876), which was the WRONG gate for this test's actual claim
* ("a node re-hits the identical condition on restart and is refused
* promotion"). The periodic-timer seam is still used, correctly, for the
* lighter-weight post-self-heal re-eligibility confirmation below. This
* test still does NOT (and cannot, from persist.c alone) prove the node's
* very FIRST-EVER promotion evaluation in a process's life is protected —
* see the promotion-gate-confirmation block's comment for the precise
* scope boundary and file:line citations.
*/
#include <rtti.h>
#include <capnp_c.h>
#include <errno.h>
#include <inttypes.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <unity/unity.h>
#include "cluster.h"
#include "controller.capnp.h"
#include "delay.h"
#include "ha.h"
#include "hook.h"
#include "jobj.h"
#include "persist.h"
#include "wal_types.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "xstring.h"
#ifdef XCE_TESTING
/* job.c's XCE_TESTING-only fault-injection seam (task 3.1) — reused
* verbatim, not reimplemented. */
extern void job_array_apply_test_set_force_fail_at(int index);
extern void job_array_apply_test_clear_force_fail(void);
/* ha.c's XCE_TESTING-only seam (ha-commit-durability task 4.4) — drives the
* exact bounded-window promotion-gate re-check without staging a full
* election. Used below ONLY for the post-self-heal re-eligibility
* confirmation (the periodic re-alert timer path, ha.c ~876) — NOT for the
* refusal proof, which must use the real election-time gate below (task 7.2
* remediation round 1: this file previously misused this seam for the
* refusal proof itself). */
extern void ha_arm_promotion_fence_timer_for_test(cluster_t *, bool,
uint64_t, uint64_t);
/* ha.c's XCE_TESTING-only seam (fix-promotion-standby-registration) — drives
* raft_cb_become_leader() (the REAL election-time promotion path) with a
* live cluster pointer, without staging a full election. This is the SAME
* seam test_convergence_verification.c's
* test_promotion_refused_when_apply_resource_fenced_even_if_eligible (task
* 4.3) already uses to prove the election-time gate refuses promotion for a
* fence engaged by the LIVE apply pump. This test drives it for a fence
* engaged by BOOT REPLAY instead (task 7.2's actual deliverable) — task 7.2
* remediation round 1: the original version of this test proved refusal via
* the WRONG gate (ha_arm_promotion_fence_timer_for_test's periodic re-alert
* timer, ha.c ~876) instead of this one. */
extern void raft_cb_become_leader_with_cluster_for_test(cluster_t *,
uint64_t);
/* jobj.c's jobsub table / 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_trf_jobobj_ready;
static cluster_t g_trf_c; /* large — must not live on the test stack */
static int trf_run_delay_task(delay_task_entry_fn_t entry, void *arg)
{
return entry(arg);
}
static void trf_build_cluster(void)
{
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32,
.delta = 16 };
kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 32,
.delta = 16 };
if (makeString((char *)"trf-probe") == NULL)
xstring_table_init(64);
if (!g_trf_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_trf_jobobj_ready = 1;
}
memset(&g_trf_c, 0, sizeof(g_trf_c));
g_trf_c.myhost = &g_trf_c.myhost_info;
g_trf_c.myhost_info.name = (char *)"ctrl-test";
g_trf_c.myhost_info.key = (char *)"";
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&g_trf_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
g_trf_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
g_trf_c.qmgr = makeInstance(&xkvmgr_type, &str_params);
g_trf_c.workermgr = makeInstance(&xkvmgr_type, &str_params);
g_trf_c.resmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(g_trf_c.jobmgr);
TEST_ASSERT_NOT_NULL(g_trf_c.qmgr);
TEST_ASSERT_NOT_NULL(g_trf_c.workermgr);
TEST_ASSERT_NOT_NULL(g_trf_c.resmgr);
zarray_init(&g_trf_c.sharedinst, 16, NULL);
}
static void trf_teardown_cluster(void)
{
zarray_fini(&g_trf_c.sharedinst);
if (g_trf_c.resmgr) dereference(g_trf_c.resmgr);
if (g_trf_c.workermgr) dereference(g_trf_c.workermgr);
if (g_trf_c.qmgr) dereference(g_trf_c.qmgr);
if (g_trf_c.jobmgr) dereference(g_trf_c.jobmgr);
xce_hook_fini(&g_trf_c.hooksys);
}
static job_t *trf_find_job(uint64_t jobid)
{
job_t *j = NULL;
lock(g_trf_c.jobmgr, RTTI_LOCK_READ);
int found = key_find(g_trf_c.jobmgr, (void *)jobid, (void **)&j);
unlock(g_trf_c.jobmgr);
return (found > 0) ? j : NULL;
}
/* Encodes a (base_jobid, array_size, spec) JobSubmit payload — the exact
* WAL_JOB_ARRAY_SUBMIT payload shape (mirrors test_job_array_fence.c's
* tf_encode_job_submit). Caller frees *out. */
static int trf_encode_job_submit(uint64_t base_jobid, uint64_t array_size,
job_submit_data_t *spec, uint8_t **out,
size_t *out_len)
{
struct capn c;
struct capn_segment *cs;
struct JobSubmit cjs;
JobSubmit_ptr jsp;
xce_job_submit_t s;
memset(&s, 0, sizeof(s));
s.jobid = base_jobid;
s.array_size = array_size;
s.submit = spec;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
encode_JobSubmit(cs, &cjs, &s);
jsp = new_JobSubmit(cs);
write_JobSubmit(&cjs, jsp);
capn_setp(capn_root(&c), 0, jsp.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;
}
/* ── The test: crash-while-would-be-fenced, restart into a boot replay that
* hits the SAME resource-exhaustion condition ────────────────────────────── */
static void
test_replay_wal_engages_fence_on_resource_exhaustion_failure(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_trf_replay_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
const uint64_t base_jobid = 980000;
const uint64_t array_size = 5;
const int fail_index = 2; /* mid-array: proves a genuine halt, not a
* lucky last-element edge case */
job_submit_data_t spec;
memset(&spec, 0, sizeof(spec));
spec.queue = (char *)"normal";
spec.user = (char *)"tester";
spec.mintasks = 1;
spec.maxtasks = 1;
spec.job_name = (char *)"replay-fence-e2e";
uint8_t *payload = NULL;
size_t len = 0;
int enc_rc = trf_encode_job_submit(base_jobid, array_size, &spec,
&payload, &len);
/* ── Phase 1 ("boot 1"): append the real committed-shape array-submit
* WAL entry physically to disk, then tear down WITHOUT ever applying
* it -- simulating a crash before the live apply pump ran (e.g. the
* process died moments after append, before any apply kick fired). No
* cluster_t is needed: persist_wal_async is a physical-write-only path
* (mirrors test_replay_apply_seam.c's phase 1). ─────────────────────── */
{
persist_config_t cfg1 = persist_config_defaults();
cfg1.wal_sync_ms = 5000;
persist_t *p1 = calloc(1, sizeof(persist_t));
TEST_ASSERT_NOT_NULL(p1);
persist_set_workdir(p1, tmp, "ctrl-test");
TEST_ASSERT_EQUAL_INT(0, persist_init(p1, &cfg1, NULL));
TEST_ASSERT_EQUAL_INT(0, persist_start(p1));
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(p1, WAL_JOB_ARRAY_SUBMIT, payload,
(uint32_t)len));
TEST_ASSERT_EQUAL_INT(0, persist_flush(p1));
persist_stop(p1);
persist_fini(p1);
free(p1);
}
/* ── Phase 2 ("boot 2" / restart): fresh cluster_t whose EMBEDDED
* persist_t points at the SAME on-disk workdir. Arm the identical
* construction-phase resource-exhaustion condition BEFORE
* persist_recover() runs, so replay_wal() hits it during REPLAY. ────── */
trf_build_cluster();
persist_config_t cfg2 = persist_config_defaults();
cfg2.wal_sync_ms = 5000;
persist_set_workdir(&g_trf_c.persist, tmp, "ctrl-test");
TEST_ASSERT_EQUAL_INT(0,
persist_init(&g_trf_c.persist, &cfg2, &g_trf_c));
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_trf_c.persist));
job_array_apply_test_clear_force_fail();
job_array_apply_test_set_force_fail_at(fail_index);
bool fenced_precondition = persist_apply_resource_fenced(&g_trf_c.persist);
int recover_rc = persist_recover(&g_trf_c.persist);
uint64_t applied_after_replay =
persist_last_applied_lsn(&g_trf_c.persist);
bool fenced_after_replay = persist_apply_resource_fenced(&g_trf_c.persist);
bool any_element_visible_after_replay = false;
for (uint64_t i = 0; i < array_size; i++) {
if (trf_find_job(base_jobid + i) != NULL)
any_element_visible_after_replay = true;
}
/* ── Promotion-gate confirmation ──────────────────────────────────────
* task 7.2 remediation (round 1): drive the REAL election-time
* eligibility gate (raft_cb_become_leader, ha.c:1303-1408 — reads
* persist_apply_resource_fenced() at ha.c:1337) directly via
* raft_cb_become_leader_with_cluster_for_test(), the SAME seam
* test_convergence_verification.c's task-4.3 sibling test already uses
* for a fence engaged by the LIVE apply pump. This is the gate an actual
* RESTART's re-election evaluates, NOT the periodic re-alert timer used
* below only for the post-self-heal confirmation.
*
* SCOPE NOTE (adversarial review finding, not fixed here — see ha.c
* comment near the fence-engage branch in persist.c's replay_wal() for
* the full citation): this only proves a promotion evaluation that runs
* AFTER a boot replay has ALREADY completed and ALREADY engaged the
* fence in this process — exactly what "boot 2" above is. It does NOT
* prove the node's very FIRST-EVER promotion evaluation in a process's
* life is protected: that check (ha.c:1303-1408) runs strictly BEFORE
* the is_first_boot block (ha.c:1427-1439) that triggers
* cluster_init_phase2()/persist_recover()/replay_wal() (cluster.c:1482)
* for the very first time, with no re-check afterward — a fence engaged
* by THAT SAME first replay is invisible to the one promotion check
* gating that boot. Closing that gap needs an ha.c change, out of this
* task's persist.c-only boundary. */
ha_config_t hcfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
};
TEST_ASSERT_EQUAL_INT(0, ha_init(&g_trf_c.ha, &hcfg, 0));
g_trf_c.ha.raft_c = &g_trf_c;
g_trf_c.ha.repl = NULL; /* no healthy peer reachable at this level */
TEST_ASSERT_EQUAL_INT(0,
delaymgr_init(&g_trf_c.delaymgr, 5, trf_run_delay_task));
TEST_ASSERT_EQUAL_INT(0, delaymgr_start(&g_trf_c.delaymgr));
/* Structurally healthy in every OTHER respect (not truncated,
* durable_floor >= promotion_safe_lsn — no elect_ctx is set here so
* promotion_safe_lsn defaults to 0 inside raft_cb_become_leader) so the
* ONLY thing that can keep the member refused is the live apply
* resource fence boot replay just engaged — same isolation rationale as
* test_job_array_fence.c's step 3 and test_convergence_verification.c's
* task 4.3 sibling test. */
g_trf_c.persist.recovery_verdict = RECOVER_CLEAN;
ha_set_promotion_fence_window_ms(60);
raft_cb_become_leader_with_cluster_for_test(&g_trf_c, /*term=*/1);
int32_t role_after_election_check =
atomic_load_explicit(&g_trf_c.ha.role, memory_order_acquire);
int in_catch_up_after_election_check =
atomic_load_explicit(&g_trf_c.ha.in_catch_up, memory_order_acquire);
bool fenced_after_promotion_check =
persist_apply_resource_fenced(&g_trf_c.persist);
/* ── Self-heal: clear the forced condition and retry via the live apply
* pump directly -- the SAME mechanism persist_recover()'s own
* post-recovery kick would eventually invoke; driving it here directly
* keeps this test deterministic (no vtask pool / async dispatch
* needed). White-box "mark committed" mirrors test_job_array_fence.c's
* own established convention for this exact fence area. ─────────────── */
job_array_apply_test_clear_force_fail();
atomic_store(&g_trf_c.persist.committed_lsn, 1); /* the one array entry */
int retry_rc = persist_apply_pump_run(&g_trf_c.persist);
uint64_t applied_after_retry = persist_last_applied_lsn(&g_trf_c.persist);
bool fenced_after_retry = persist_apply_resource_fenced(&g_trf_c.persist);
bool all_elements_visible_after_retry = true;
for (uint64_t i = 0; i < array_size; i++) {
if (trf_find_job(base_jobid + i) == NULL)
all_elements_visible_after_retry = false;
}
/* Post-self-heal re-eligibility confirmation uses the SIBLING periodic
* re-alert timer gate (ha.c ~876) rather than re-driving
* raft_cb_become_leader_with_cluster_for_test() a second time: both
* gates read the exact same persist_apply_resource_fenced() flag this
* fixture cleared above, so this is a legitimate, lighter-weight proof
* that clearing the fence (regardless of which engaged it) restores
* eligibility everywhere it is checked -- not a re-use of the "wrong"
* gate this remediation round fixed above (that fix was specifically
* about which gate proves the REFUSAL, not this self-heal check). */
ha_arm_promotion_fence_timer_for_test(&g_trf_c, /*truncated=*/false,
/*durable_floor=*/0, /*hwm=*/0);
usleep(250 * 1000);
int in_catch_up_after_self_heal =
atomic_load_explicit(&g_trf_c.ha.in_catch_up, memory_order_acquire);
ha_set_promotion_fence_window_ms(0); /* restore default */
ha_cancel_promotion_fence_timer(&g_trf_c.ha);
delaymgr_stop(&g_trf_c.delaymgr);
delaymgr_fini(&g_trf_c.delaymgr);
ha_fini(&g_trf_c.ha);
job_array_apply_test_clear_force_fail();
free(payload);
persist_stop(&g_trf_c.persist);
persist_fini(&g_trf_c.persist);
trf_teardown_cluster();
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);
/* ── Assertions ───────────────────────────────────────────────────── */
TEST_ASSERT_EQUAL_INT(0, enc_rc);
TEST_ASSERT_FALSE_MESSAGE(fenced_precondition,
"a fresh persist_t must start with the Live Apply Resource Fence "
"CLEAR");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, recover_rc,
"a resource-exhaustion apply failure during boot replay must NOT "
"fail the boot itself (Req 9.4: never treat this as a reason to "
"fail the commit)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, applied_after_replay,
"last_applied_lsn must NOT advance past the failed array LSN during "
"boot replay -- mirrors persist_apply_pump_run's live-path halt");
TEST_ASSERT_TRUE_MESSAGE(fenced_after_replay,
"a node that crashes and restarts into a boot replay hitting the "
"IDENTICAL resource-exhaustion condition must re-engage the Live "
"Apply Resource Fence DURING REPLAY (task 4.2's self-flagged gap, "
"closed by task 7.2) -- not silently boot clean and look promotable "
"while actually divergent/behind");
TEST_ASSERT_FALSE_MESSAGE(any_element_visible_after_replay,
"no partial/divergent set of array elements may be constructed "
"during a fenced boot replay (Req 9.6)");
TEST_ASSERT_NOT_EQUAL_MESSAGE((int32_t)CTRL_PRIMARY,
role_after_election_check,
"a node fenced by BOOT REPLAY must be refused promotion by the REAL "
"election-time eligibility gate (raft_cb_become_leader, ha.c:1303-"
"1408) -- the same gate tasks 4.2/4.3 already wired for a live-apply-"
"engaged fence, now proven for a boot-replay-engaged one too (task "
"7.2 remediation round 1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, in_catch_up_after_election_check,
"a node fenced by BOOT REPLAY must be refused promotion by the SAME "
"existing ha.c gate (tasks 4.2/4.3) a live-apply fence already "
"uses -- no separate mechanism, no drift");
TEST_ASSERT_TRUE_MESSAGE(fenced_after_promotion_check,
"the fence must still read engaged after the promotion-gate check "
"-- only the apply pump's own successful retry may clear it");
TEST_ASSERT_TRUE_MESSAGE(retry_rc >= 0,
"once the forced condition clears, retrying the SAME stuck LSN via "
"the live apply pump must succeed");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, applied_after_retry,
"last_applied_lsn must advance to the array LSN once the retry "
"succeeds");
TEST_ASSERT_TRUE_MESSAGE(all_elements_visible_after_retry,
"every array element must be visible once the retry succeeds");
TEST_ASSERT_FALSE_MESSAGE(fenced_after_retry,
"the Live Apply Resource Fence must self-clear once the retry "
"succeeds (Req 9.5), regardless of whether it was engaged by boot "
"replay or the live apply pump");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, in_catch_up_after_self_heal,
"once the fence clears, the node must become eligible for "
"promotion again with no other action");
printf("test_replay_wal_engages_fence_on_resource_exhaustion_failure: "
"PASS\n");
}
#endif /* XCE_TESTING */
void test_job_array_replay_fence(void)
{
#ifdef XCE_TESTING
RUN_TEST(test_replay_wal_engages_fence_on_resource_exhaustion_failure);
#else
printf("test_job_array_replay_fence: skipped (no XCE_TESTING)\n");
#endif
}