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

fix(job-array-single-wal-record): replay_wal's job-count tally was blind to array submissions

Browse source 
replay_wal() (persist.c ~3902) tallied *new_job_submits_out only for
hdr.type == WAL_JOB_SUBMIT. WAL_JOB_ARRAY_SUBMIT entries were
dispatched and correctly materialized N real jobmgr entries via
apply_committed_entry()'s array case, but never counted --
persist_verify_apply_correctness() then saw real jobmgr growth exceed
the undercounted expectation and declared APPLY_DIVERGENT, engaging a
PERMANENT, non-self-healing fence (distinct from task 4.2's
self-healing Live Apply Resource Fence).

Live-reproduced during task 6.1's soak attempt on 2 of 3 nodes in a
single small-array (N=2) sub-round via two different trigger paths.
Deterministic, not size-dependent -- the first array submission to a
live cluster would fence any node that restarts or hits an
election-path recovery fallback before its next checkpoint, shrinking
the promotable pool. A pre-existing comment at persist.c ~3863-3874
(predating this spec) had already flagged this exact code region as
needing re-verification before tasks 4.1/6.1+ went live; this is that
re-verification, and it failed.

Fix adds a WAL_JOB_ARRAY_SUBMIT branch tallying apply_rc (the genuine
per-record element count apply_committed_entry()'s array case already
returns) when apply_rc > 0, mirroring the existing dedup-safe pattern
for WAL_JOB_SUBMIT. Traced the full return-value space of
restore_job_array()/apply_committed_entry()'s array case to confirm
every tallied value is provably equal to real jobmgr growth -- the fix
cannot overcount and so cannot mask a genuine divergence.

Verified via a from-scratch live 3-node HA soak re-run (--array-soak),
independently by both implementer and reviewer: zero APPLY_DIVERGENT
across all three original trigger paths, one-WAL-record/one-commit-
barrier confirmed for N=2 and N=10000, per-element post-failover
status/cancel/completion confirmed.

Task 6.0a of job-array-single-wal-record (Requirements 3.1, 3.2, 4.1,
9.4). Inserted mid-run after live soak testing surfaced the bug; user
confirmed sequencing (fix now, blocking task 6.1+).
This commit is contained in:
Rongsong Shen 2026-07-16 04:09:43 +00:00
parent b7b90e9d21
commit 16747aa96d
3 changed files with 292 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

24
src/controller/persist.c
View file

@ -3903,6 +3903,30 @@ static int replay_wal(persist_t* p, uint64_t from_lsn,
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;
/* Advance current_lsn to one past this entry */
if (hdr.lsn + 1 > p->current_lsn)
p->current_lsn = hdr.lsn + 1;

257
src/controller/testsuite/test_apply_correctness_verification.c
View file

@ -131,6 +131,43 @@ static int tacv_encode_job(const xce_job_t* j, uint8_t** out, size_t* out_len)
return 0;
}
/* Encodes a (base_jobid, array_size, spec) JobSubmit payload -- the exact
* shape a committed WAL_JOB_ARRAY_SUBMIT record carries (mirrors
* test_job_array_apply.c's taj_encode_job_submit and persist.c's
* persist_submit_job_array_async encode side). Caller frees *out. */
static int tacv_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;
}
static persist_t* tacv_make_fresh_persist(const char* tmpdir)
{
persist_t* p = calloc(1, sizeof(persist_t));
@ -720,6 +757,224 @@ static void test_real_replay_duplicate_jobid_is_apply_clean_no_fence(void)
printf("test_real_replay_duplicate_jobid_is_apply_clean_no_fence: PASS\n");
}
/*
* job-array-single-wal-record task 6.0a (Req 3.1, 3.2, 4.1, 9.4): a
* committed WAL_JOB_ARRAY_SUBMIT record's real N-element jobmgr growth must
* be counted in replay_wal()'s new_job_submits_out tally, mirroring the
* WAL_JOB_SUBMIT tally above. Before the fix, replay_wal() only checked
* hdr.type == WAL_JOB_SUBMIT, so an array record's genuine N-element growth
* (apply_committed_entry()'s WAL_JOB_ARRAY_SUBMIT case correctly returns
* apply_rc == N) contributed 0 to the tally -- persist_verify_apply_
* correctness() then compared real growth (N) against the undercounted
* expectation (0), producing a false APPLY_DIVERGENT verdict that engages
* the PERMANENT, non-self-healing persist_recover_loud_fence_if_
* unrecoverable() fence (distinct from task 4.2's self-healing Live Apply
* Resource Fence). Live-reproduced during task 6.1's soak (2026-07-15) on 2
* of 3 nodes after a small (N=2) array submit followed by a node restart.
*
* This test is the exact live-soak reproduction: an ordinary WAL_JOB_SUBMIT
* followed by a small WAL_JOB_ARRAY_SUBMIT, replayed on restart. A correct
* tally must report APPLY_CLEAN and must never trip the permanent
* apply-divergent fence; every array element must be genuinely
* materialized.
*/
static void
test_real_replay_array_submit_tallies_correctly_no_false_divergence(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_tacv_array_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
const uint64_t base_jobid = 81000;
const uint64_t array_size = 2;
/* Phase 1: one ordinary WAL_JOB_SUBMIT, then one small WAL_JOB_ARRAY_
* SUBMIT -- mirrors the live soak's mixed ordinary-submit + array-submit
* WAL, replayed whole on restart. */
{
persist_t* p = tacv_make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p);
xce_job_t jdef;
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = 80999;
jdef.state = JOB_STAT_PEND;
job_submit_data_t plain_submit;
memset(&plain_submit, 0, sizeof(plain_submit));
plain_submit.queue = (char*)"rq1";
plain_submit.user = (char*)"tester";
jdef.submit = &plain_submit;
uint8_t* plain_payload = NULL;
size_t plain_len = 0;
TEST_ASSERT_EQUAL_INT(0,
tacv_encode_job(&jdef, &plain_payload, &plain_len));
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(p, WAL_JOB_SUBMIT, plain_payload,
(uint32_t)plain_len));
free(plain_payload);
job_submit_data_t array_spec;
memset(&array_spec, 0, sizeof(array_spec));
array_spec.queue = (char*)"rq1";
array_spec.user = (char*)"tester";
uint8_t* array_payload = NULL;
size_t array_len = 0;
TEST_ASSERT_EQUAL_INT(0,
tacv_encode_job_submit(base_jobid, array_size, &array_spec,
&array_payload, &array_len));
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(p, WAL_JOB_ARRAY_SUBMIT, array_payload,
(uint32_t)array_len));
free(array_payload);
TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
tacv_teardown_persist(p);
}
/* Phase 2: fresh persist_t + fresh cluster_t, recover from disk. */
{
persist_t* p2 = tacv_make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p2);
tacv_build_cluster();
p2->cluster = &g_tacv_c;
TEST_ASSERT_EQUAL_INT(0, persist_recover(p2));
TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN,
persist_apply_verdict(p2),
"task 6.0a: a WAL_JOB_ARRAY_SUBMIT's real N-element jobmgr "
"growth must be counted in replay_wal's tally -- an "
"undercounted tally must NOT produce a false APPLY_DIVERGENT "
"verdict");
TEST_ASSERT_FALSE_MESSAGE(
persist_recovery_unrecoverable(p2),
"task 6.0a: a correctly-tallied array replay must never trip "
"the permanent, non-self-healing recovery_unrecoverable fence");
job_t* j0 = NULL;
job_t* j1 = NULL;
lock(g_tacv_c.jobmgr, RTTI_LOCK_READ);
key_find(g_tacv_c.jobmgr, (void*)(base_jobid + 0), (void**)&j0);
key_find(g_tacv_c.jobmgr, (void*)(base_jobid + 1), (void**)&j1);
unlock(g_tacv_c.jobmgr);
TEST_ASSERT_NOT_NULL_MESSAGE(j0,
"array element 0 must be materialized");
TEST_ASSERT_NOT_NULL_MESSAGE(j1,
"array element 1 must be materialized");
tacv_teardown_cluster();
tacv_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_real_replay_array_submit_tallies_correctly_no_false_"
"divergence: PASS\n");
}
/*
* Idempotency (mirrors test_real_replay_duplicate_jobid_is_apply_clean_
* no_fence's single-job pattern, extended to a whole array as one unit --
* bug-1419's array analog, the trigger task 6.2's future soak scenario
* exercises at the wire level): the SAME array record committed twice as two
* separate WAL_JOB_ARRAY_SUBMIT entries (a commit-timeout retry whose
* original entry also later achieves commit) must replay APPLY_CLEAN.
* restore_job_array()'s per-index idempotency check materializes the array
* only once (first application: apply_rc == array_size; second application:
* every index already exists -> restore_job_array() returns -EEXIST ->
* apply_committed_entry() returns 0) -- replay_wal's corrected tally must
* agree with jobmgr's real growth of exactly array_size, not
* 2 * array_size.
*/
static void
test_real_replay_duplicate_array_submit_is_apply_clean_no_fence(void)
{
char tmp[256];
strcpy(tmp, "/tmp/xce_tacv_array_dup_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
const uint64_t base_jobid = 82000;
const uint64_t array_size = 3;
/* Phase 1: append the SAME array record twice as two separate
* WAL_JOB_ARRAY_SUBMIT entries. */
{
persist_t* p = tacv_make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p);
for (int attempt = 0; attempt < 2; attempt++) {
job_submit_data_t spec;
memset(&spec, 0, sizeof(spec));
spec.queue = (char*)"rq1";
spec.user = (char*)"tester";
uint8_t* payload = NULL;
size_t len = 0;
TEST_ASSERT_EQUAL_INT(0,
tacv_encode_job_submit(base_jobid, array_size, &spec,
&payload, &len));
TEST_ASSERT_EQUAL_INT(0,
persist_wal_async(p, WAL_JOB_ARRAY_SUBMIT, payload,
(uint32_t)len));
free(payload);
}
TEST_ASSERT_EQUAL_INT(0, persist_flush(p));
tacv_teardown_persist(p);
}
/* Phase 2: fresh persist_t + fresh cluster_t, recover from disk. */
{
persist_t* p2 = tacv_make_fresh_persist(tmp);
TEST_ASSERT_NOT_NULL(p2);
tacv_build_cluster();
p2->cluster = &g_tacv_c;
TEST_ASSERT_EQUAL_INT(0, persist_recover(p2));
TEST_ASSERT_EQUAL_INT_MESSAGE(APPLY_CLEAN,
persist_apply_verdict(p2),
"task 6.0a idempotency: a retried (duplicate) array commit "
"must replay APPLY_CLEAN -- the second, idempotent "
"whole-record re-apply (-EEXIST, apply_rc==0) must not be "
"double-counted into the tally");
TEST_ASSERT_FALSE_MESSAGE(
persist_recovery_unrecoverable(p2),
"a duplicate array commit must never trip the permanent "
"apply-divergent fence");
for (uint64_t i = 0; i < array_size; i++) {
job_t* j = NULL;
lock(g_tacv_c.jobmgr, RTTI_LOCK_READ);
key_find(g_tacv_c.jobmgr, (void*)(base_jobid + i), (void**)&j);
unlock(g_tacv_c.jobmgr);
TEST_ASSERT_NOT_NULL_MESSAGE(j,
"every array index must be materialized exactly once");
}
tacv_teardown_cluster();
tacv_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_real_replay_duplicate_array_submit_is_apply_clean_no_"
"fence: PASS\n");
}
/*
* raft-apply-decoupling task 5.2 round 2 (Req 8.3 remediation): the
* apply_divergent ERROR log must NOT append the shared structural-fence
@ -1248,6 +1503,8 @@ void test_apply_correctness_verification(void)
RUN_TEST(test_apply_divergent_decision_forces_unconditional_fence);
RUN_TEST(test_apply_divergent_loud_fence_sets_sticky_flag_and_stays_alive);
RUN_TEST(test_real_replay_duplicate_jobid_is_apply_clean_no_fence);
RUN_TEST(test_real_replay_array_submit_tallies_correctly_no_false_divergence);
RUN_TEST(test_real_replay_duplicate_array_submit_is_apply_clean_no_fence);
RUN_TEST(test_apply_divergent_loud_fence_log_omits_false_peer_hwm_clause);
RUN_TEST(test_apply_divergent_process_stays_alive_and_functional_after_fence);
RUN_TEST(test_apply_divergent_fence_clears_via_existing_clear_mechanism);