/* test_job_array_fence.c — job-array-single-wal-record task 4.4 * * Integration capstone for the Live Apply Resource Fence (design.md "Live * Apply Resource Fence"; Requirements 9.4, 9.5, 9.6). This is the ONE test * that drives the FULL engage-refuse-clear-recover sequence end-to-end * through the real production seams built by tasks 3.1/4.1/4.2/4.3 — it adds * no new mechanism, it only wires the existing pieces together the way a * live cluster would exercise them: * * 1. Force a resource-exhaustion-style failure during array expansion. * 2. The node fences itself (persist_apply_resource_fenced() reads true). * 3. The node is refused promotion (ha.c's periodic fence-release check — * task 4.3's site 1, ha_arm_promotion_fence_timer_for_test). * 4. No partial or divergent set of elements was constructed. * 5. The failure does not affect another (simulated) node's independent * progress. * 6. Clearing the forced condition and retrying succeeds: the fence * self-clears and the node is eligible for promotion again. * * ── Seam choice (construction-phase, not bulk-insert-phase) ──────────────── * job.c exposes TWO distinct XCE_TESTING fault-injection points for * restore_job_array() (see tasks.md Implementation Notes, task 3.1): * - job_array_apply_test_set_force_fail_at(idx): fails the SCRATCH-BUILD * phase (makeJob() at index idx). Proven fully atomic: nothing is ever * written to c->jobmgr on this path (test_job_array_apply.c, test_apply_ * pump.c tests 2b-2d). * - job_array_apply_test_set_bulk_insert_force_fail_at(idx): fails the * final BULK-INSERT pass (key_add() at index idx). tasks.md's task 3.1 * Implementation Note documents this as a NARROWER, NON-atomic case by * design: indices [0, idx) are already genuinely, permanently inserted * into c->jobmgr (no key_del exists to undo them) before the failure is * detected — proven directly by test_rja_bulk_insert_failure_reports_ * error_not_false_success in test_job_array_apply.c. * Requirement 9.6 (and this task's own step 4) requires proving "constructs * no partial or divergent set of elements" — a claim that is TRUE for the * construction-phase seam and, by the task 3.1 Implementation Note's own * documented residual gap, FALSE for the bulk-insert-phase seam. Using the * bulk-insert seam here would either make this test's own step-4 assertion * fail or force weakening it to hide a known gap — neither is acceptable for * the capstone test that is supposed to PROVE Requirement 9.6. This file * therefore drives the construction-phase seam for the full sequence. The * bulk-insert-phase seam's own (already known, already tested at the * restore_job_array() level) residual gap is a separate, pre-existing, * documented finding — not something this task's Observable asks it to * close. * * ── Fixture ────────────────────────────────────────────────────────────── * A single, real cluster_t (g_tf_c) combining: * - test_apply_pump.c's tap_setup shape (real persist_init/persist_start, * jobmgr/qmgr/workermgr/resmgr, hooksys, a wired commit engine with two * stub standbys) — needed to append a real WAL_JOB_ARRAY_SUBMIT entry * and drive persist_apply_pump_run()/restore_job_array() for real. * - test_commit_durability.c's dur_make_fence_cluster shape (ha_init + * a real, started delaymgr) — needed to drive the SAME cluster_t's * promotion-fence gate (ha_arm_promotion_fence_timer_for_test, task * 4.3's site 1) through a real bounded-window timer. * The commit path itself is driven white-box (persist_wal_async + * atomic_store(committed_lsn) + a direct persist_apply_pump_run() call) — * the exact pattern test_apply_pump.c's own task-4.1/4.2 tests * (test_apply_pump_run_halts_at_failed_lsn_without_advancing, * test_apply_pump_run_engages_live_fence_on_failed_lsn, * test_apply_pump_run_self_clears_fence_on_successful_retry) already use for * this exact WAL_JOB_ARRAY_SUBMIT/fence area — reused rather than driving * the full multi-node repl_on_ack commit-engine path, which would add * flakiness risk (see that file's own header comment on the fsync-thread * race) without adding proof of anything this test needs: the seam under * test is restore_job_array()'s construction failure and persist_apply_pump_ * run()'s halt/fence/self-heal, which behave identically once committed_lsn * advances, however it advanced. * * Step 5 ("does not affect any other node's independent progress") uses a * second, wholly separate cluster_t (node2, jobmgr-only, mirroring test_job_ * array_apply.c's taj_setup) and calls restore_job_array() on it directly — * proving the fence state lives on g_tf_c's OWN persist_t (a parameter, not * global state) and that a different cluster_t's independent apply succeeds * untouched by g_tf_c's fenced condition. The XCE_TESTING fault-injection * variables ARE process-global (job.c's own doc comment: "a later task * needing independent multi-node fault injection can revisit this"), so this * is the lighter-weight, sequential proof the task description explicitly * sanctions rather than a true concurrent multi-process harness. */ #include #include #include #include #include #include #include #include #include #include "cluster.h" #include "commit.h" #include "controller.capnp.h" #include "delay.h" #include "ha.h" #include "hook.h" #include "jobj.h" #include "persist.h" #include "repl.h" #include "wal_types.h" #include "xce/cluster_def.h" #include "xce/types.h" #include "xlist.h" #include "xstring.h" #ifdef XCE_TESTING /* 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_tf_jobobj_ready; static cluster_t g_tf_c; /* large — must not live on the test stack */ static commit_engine_t g_tf_hce; static int g_tf_hep_a, g_tf_hep_b; /* dummy standby endpoints */ static char g_tf_tmpdir[256]; extern int (*repl_broadcast_send_hook)(void *ep, wal_catchup_entry_t **entries, int n_entries); static int tf_drop_send_hook(void *ep, wal_catchup_entry_t **entries, int n_entries) { (void)ep; (void)entries; (void)n_entries; return 0; } /* The delaymgr's task-run wrapper — a trivial synchronous pass-through * (mirrors test_commit_durability.c's dur_run_delay_task); the real * delaymgr worker thread still owns the actual wall-clock scheduling, this * is only how each fired task is invoked. */ static int tf_run_delay_task(delay_task_entry_fn_t entry, void *arg) { return entry(arg); } static void tf_setup(void) { kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING, .nslots = 32, .delta = 16 }; kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; if (makeString((char *)"tf-probe") == NULL) xstring_table_init(64); if (!g_tf_jobobj_ready) { TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32)); g_tf_jobobj_ready = 1; } memset(&g_tf_c, 0, sizeof(g_tf_c)); g_tf_c.myhost = &g_tf_c.myhost_info; g_tf_c.myhost_info.name = (char *)"ctrl-test"; g_tf_c.myhost_info.key = (char *)""; TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&g_tf_c.hooksys, XCE_CONTROLLER_HOOK_MAX)); g_tf_c.jobmgr = makeInstance(&xkvmgr_type, &int_params); g_tf_c.qmgr = makeInstance(&xkvmgr_type, &str_params); g_tf_c.workermgr = makeInstance(&xkvmgr_type, &str_params); g_tf_c.resmgr = makeInstance(&xkvmgr_type, &str_params); TEST_ASSERT_NOT_NULL(g_tf_c.jobmgr); TEST_ASSERT_NOT_NULL(g_tf_c.qmgr); TEST_ASSERT_NOT_NULL(g_tf_c.workermgr); TEST_ASSERT_NOT_NULL(g_tf_c.resmgr); zarray_init(&g_tf_c.sharedinst, 16, NULL); persist_config_t cfg = persist_config_defaults(); cfg.wal_sync_ms = 5000; cfg.sync_timeout_ms = 2000; snprintf(g_tf_tmpdir, sizeof(g_tf_tmpdir), "/tmp/xce_tf44_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(g_tf_tmpdir)); persist_set_workdir(&g_tf_c.persist, g_tf_tmpdir, "ctrl"); TEST_ASSERT_EQUAL_INT(0, persist_init(&g_tf_c.persist, &cfg, &g_tf_c)); TEST_ASSERT_EQUAL_INT(0, persist_start(&g_tf_c.persist)); g_tf_c.persist.recovery_verdict = RECOVER_CLEAN; TEST_ASSERT_TRUE( repl_add_standby(&g_tf_c.persist.repl, &g_tf_hep_a, "sb_a", 7500) >= 0); TEST_ASSERT_TRUE( repl_add_standby(&g_tf_c.persist.repl, &g_tf_hep_b, "sb_b", 7501) >= 0); TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_tf_hce, &g_tf_c.persist, 3, 8)); repl_set_commit_engine(&g_tf_c.persist.repl, &g_tf_hce); repl_broadcast_send_hook = tf_drop_send_hook; 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_tf_c.ha, &hcfg, 0)); g_tf_c.ha.raft_c = &g_tf_c; g_tf_c.ha.repl = NULL; /* no healthy peer reachable at this level */ TEST_ASSERT_EQUAL_INT(0, delaymgr_init(&g_tf_c.delaymgr, 5, tf_run_delay_task)); TEST_ASSERT_EQUAL_INT(0, delaymgr_start(&g_tf_c.delaymgr)); job_array_apply_test_clear_force_fail(); job_array_apply_test_clear_bulk_insert_force_fail(); } static void tf_teardown(void) { job_array_apply_test_clear_force_fail(); job_array_apply_test_clear_bulk_insert_force_fail(); ha_cancel_promotion_fence_timer(&g_tf_c.ha); delaymgr_stop(&g_tf_c.delaymgr); delaymgr_fini(&g_tf_c.delaymgr); ha_fini(&g_tf_c.ha); repl_broadcast_send_hook = NULL; commit_engine_fini(&g_tf_hce); persist_stop(&g_tf_c.persist); persist_fini(&g_tf_c.persist); zarray_fini(&g_tf_c.sharedinst); if (g_tf_c.resmgr) dereference(g_tf_c.resmgr); if (g_tf_c.workermgr) dereference(g_tf_c.workermgr); if (g_tf_c.qmgr) dereference(g_tf_c.qmgr); if (g_tf_c.jobmgr) dereference(g_tf_c.jobmgr); xce_hook_fini(&g_tf_c.hooksys); } static job_t *tf_find_job_in(cluster_t *c, uint64_t jobid) { job_t *j = NULL; lock(c->jobmgr, RTTI_LOCK_READ); int found = key_find(c->jobmgr, (void *)jobid, (void **)&j); unlock(c->jobmgr); return (found > 0) ? j : NULL; } static job_t *tf_find_job(uint64_t jobid) { return tf_find_job_in(&g_tf_c, jobid); } static job_submit_data_t tf_make_spec(void) { job_submit_data_t s; memset(&s, 0, sizeof(s)); s.queue = (char *)"normal"; s.user = (char *)"tester"; s.mintasks = 1; s.maxtasks = 1; s.job_name = (char *)"array-fence-e2e"; return s; } /* Encodes a (base_jobid, array_size, spec) JobSubmit payload — the exact * WAL_JOB_ARRAY_SUBMIT payload shape (mirrors test_apply_pump.c's * tap_encode_job_submit / test_job_array_apply.c's taj_encode_job_submit). * Caller frees *out. */ static int tf_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 single capstone test: engage → refuse → no-partial-set → * other-node-unaffected → clear+retry → self-heal → eligible again ──────── */ static void test_job_array_fence_engage_refuse_clear_recover_end_to_end(void) { tf_setup(); const uint64_t base_jobid = 970000; 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 = tf_make_spec(); uint8_t *payload = NULL; size_t len = 0; int enc_rc = tf_encode_job_submit(base_jobid, array_size, &spec, &payload, &len); bool fenced_precondition = persist_apply_resource_fenced(&g_tf_c.persist); /* ── Step 1: force a resource-exhaustion-style failure during array * expansion (construction-phase seam — see file header). ──────────── */ job_array_apply_test_set_force_fail_at(fail_index); int rc_array = persist_wal_async(&g_tf_c.persist, WAL_JOB_ARRAY_SUBMIT, payload, (uint32_t)len); uint64_t array_lsn = g_tf_c.persist.current_lsn - 1; uint64_t baseline_applied = persist_last_applied_lsn(&g_tf_c.persist); /* White-box commit (established test_apply_pump.c pattern for this exact * WAL_JOB_ARRAY_SUBMIT/fence area — see file header). */ atomic_store(&g_tf_c.persist.committed_lsn, array_lsn); int walk_rc = persist_apply_pump_run(&g_tf_c.persist); uint64_t applied_after_failure = persist_last_applied_lsn(&g_tf_c.persist); bool fenced_after_failure = persist_apply_resource_fenced(&g_tf_c.persist); /* ── Step 4: no partial or divergent set of elements was constructed — * checked now, before anything else touches g_tf_c's jobmgr. ──────── */ bool any_element_visible_after_failure = false; for (uint64_t i = 0; i < array_size; i++) { if (tf_find_job(base_jobid + i) != NULL) any_element_visible_after_failure = true; } /* ── Step 3: refused promotion (task 4.3 site 1 — the periodic * fence-release check, ha_arm_promotion_fence_timer_for_test). The * structural conditions are made fully healthy (not truncated, * durable_floor >= hwm) so the ONLY thing that can keep the member * fenced is the live apply resource fence just engaged above. * * hwm=0 is deliberate, not a placeholder: this fixture runs persist * with a REAL background fsync worker (needed for the real WAL append * above), so durable_lsn is genuinely, monotonically advancing on its * own schedule — any fixed non-zero value this test tried to inject * white-box would race the real flusher and could be clobbered before * promotion_fence_timeout_fn re-reads it at fire time (observed * directly during development of this test). durable_lsn is monotonic * and never negative, so hwm=0 makes "durable_floor >= hwm" true by * construction at both arm time and fire time, regardless of the real * flusher's timing — isolating the ONLY variable this test cares about * (the live apply resource fence) exactly as the sibling precedent * tests (test_commit_durability.c, test_convergence_verification.c) * isolate it, just without relying on a fixture with no live WAL * activity to keep an injected value stable. */ g_tf_c.persist.recovery_verdict = RECOVER_CLEAN; ha_set_promotion_fence_window_ms(60); /* short window so the test is fast */ extern void ha_arm_promotion_fence_timer_for_test(cluster_t *, bool, uint64_t, uint64_t); ha_arm_promotion_fence_timer_for_test(&g_tf_c, /*truncated=*/false, /*durable_floor=*/0, /*hwm=*/0); int in_catch_up_on_arm = atomic_load_explicit(&g_tf_c.ha.in_catch_up, memory_order_acquire); usleep(250 * 1000); /* let the bounded re-alert window elapse and fire */ int in_catch_up_after_window = atomic_load_explicit(&g_tf_c.ha.in_catch_up, memory_order_acquire); bool fenced_after_promotion_check = persist_apply_resource_fenced(&g_tf_c.persist); /* ── Step 5: the failure does not affect another node's independent * progress. Clear the (process-global, per job.c's own doc comment) * fault-injection seam, then apply the SAME committed array payload on a * wholly separate cluster_t — g_tf_c's own fence state must be * untouched by that node's independent success. ────────────────────── */ job_array_apply_test_clear_force_fail(); cluster_t node2; memset(&node2, 0, sizeof(node2)); kvmgr_params_t node2_int_params = { .key_type = KVMGR_KEY_INT, .nslots = 32, .delta = 16 }; node2.jobmgr = makeInstance(&xkvmgr_type, &node2_int_params); TEST_ASSERT_NOT_NULL(node2.jobmgr); xlist_t node2_new; memset(&node2_new, 0, sizeof(node2_new)); int node2_rc = restore_job_array(&node2, payload, len, /*entry_lsn=*/1, &node2_new); bool node2_all_visible = true; for (uint64_t i = 0; i < array_size; i++) { if (tf_find_job_in(&node2, base_jobid + i) == NULL) node2_all_visible = false; } bool node1_still_fenced_after_node2_progress = persist_apply_resource_fenced(&g_tf_c.persist); xlist_fini(&node2_new); dereference(node2.jobmgr); /* ── Step 6: clear the forced condition and retry — the fence self-heals * and the apply completes; then re-check promotion eligibility. ────── */ int walk_rc_retry = persist_apply_pump_run(&g_tf_c.persist); uint64_t applied_after_retry = persist_last_applied_lsn(&g_tf_c.persist); bool fenced_after_retry = persist_apply_resource_fenced(&g_tf_c.persist); bool all_elements_visible_after_retry = true; for (uint64_t i = 0; i < array_size; i++) { if (tf_find_job(base_jobid + i) == NULL) all_elements_visible_after_retry = false; } ha_arm_promotion_fence_timer_for_test(&g_tf_c, /*truncated=*/false, /*durable_floor=*/0, /*hwm=*/0); usleep(250 * 1000); int in_catch_up_after_self_heal = atomic_load_explicit(&g_tf_c.ha.in_catch_up, memory_order_acquire); ha_set_promotion_fence_window_ms(0); /* restore default */ free(payload); tf_teardown(); /* ── Assertions ───────────────────────────────────────────────────── */ TEST_ASSERT_EQUAL_INT(0, enc_rc); TEST_ASSERT_EQUAL_INT(0, rc_array); TEST_ASSERT_FALSE_MESSAGE(fenced_precondition, "a fresh persist_t must start with the Live Apply Resource Fence " "CLEAR"); /* Step 1/2: forced failure genuinely halts the walk and engages the * fence. */ TEST_ASSERT_TRUE_MESSAGE(walk_rc < 0, "the forced construction-phase failure must halt the apply walk " "with a negative return, not silently succeed"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(baseline_applied, applied_after_failure, "last_applied_lsn must NOT advance past the failed array LSN"); TEST_ASSERT_TRUE_MESSAGE(fenced_after_failure, "the Live Apply Resource Fence must engage the moment the apply " "walk halts on the forced resource-exhaustion failure (Req 9.4)"); /* Step 4: no partial/divergent set of elements. */ TEST_ASSERT_FALSE_MESSAGE(any_element_visible_after_failure, "restore_job_array's scratch-build atomicity must leave ZERO array " "elements visible after a construction-phase failure — no partial " "or divergent set (Req 9.6)"); /* Step 3: refused promotion, structurally-healthy-but-for-the-fence. */ TEST_ASSERT_EQUAL_INT_MESSAGE(1, in_catch_up_on_arm, "arming the promotion fence timer must immediately fence the " "member (in_catch_up=1)"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, in_catch_up_after_window, "a node with the Live Apply Resource Fence engaged must remain " "refused for promotion even though every OTHER structural " "condition (not truncated, durable_floor >= hwm) is healthy — the " "periodic fence-release check must not clear it (Req 9.4/9.5)"); TEST_ASSERT_TRUE_MESSAGE(fenced_after_promotion_check, "the fence itself must still read engaged after the promotion " "gate check — only the apply pump's own successful retry may " "clear it"); /* Step 5: another node's independent progress is unaffected. */ TEST_ASSERT_EQUAL_INT_MESSAGE((int)array_size, node2_rc, "a wholly separate node applying the SAME committed array record " "(with the process-global fault-injection seam cleared) must " "construct the full set independently of node 1's fenced state"); TEST_ASSERT_TRUE_MESSAGE(node2_all_visible, "node 2's independent apply must produce every array element"); TEST_ASSERT_TRUE_MESSAGE(node1_still_fenced_after_node2_progress, "node 1's own fence (its own persist_t's state) must be completely " "unaffected by node 2's independent, unrelated progress (Req 9.5)"); /* Step 6: clear + retry + self-heal + eligible again. */ TEST_ASSERT_TRUE_MESSAGE(walk_rc_retry >= 0, "once the forced condition is removed, the retry of the SAME " "stuck LSN must succeed"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(array_lsn, 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, from inside the " "apply walk itself, the instant the retry of the same stuck LSN " "succeeds — no external/admin action (Req 9.5)"); 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 (task 4.3's Observable " "clause)"); printf("test_job_array_fence_engage_refuse_clear_recover_end_to_end: " "PASS\n"); } #endif /* XCE_TESTING */ void test_job_array_fence(void) { #ifdef XCE_TESTING RUN_TEST(test_job_array_fence_engage_refuse_clear_recover_end_to_end); #else printf("test_job_array_fence: skipped (no XCE_TESTING)\n"); #endif }