Rongsong Shen
b47cc660ba
Adds a new atomic LSN field on persist_t (distinct from the existing boot-only unrecoverable-fence flag) plus three primitives: persist_apply_resource_fenced(), persist_mark_apply_resource_fenced(), persist_clear_apply_resource_fence(). Wired into persist_apply_pump_run(): engages on task 4.1's halt condition, clears unconditionally on the walk's next successful advance — provably equivalent to "clears only when the same stuck LSN's retry succeeds" since the walk is strictly sequential and never advances past a failed position. Known residual gap, out of this task's boundary (documented in tasks.md Implementation Notes): replay_wal() during boot/restart replay does not yet share this protection, so a node crashing while fenced can silently re-pass the same failure on restart. Flagged for a follow-up task, not fixed here. Task 4.2 of job-array-single-wal-record (Requirements 9.4, 9.5).
1654 lines
72 KiB
C
1654 lines
72 KiB
C
/* test_apply_pump.c — raft-apply-decoupling task 6.1
|
|
*
|
|
* Proves the ordered apply pump: the strict-order walk that drives
|
|
* apply_committed_entry() (task 2.2) off live commit-advance for the very
|
|
* first time — before this task, only restart replay (task 2.3) ever called
|
|
* it. Covers:
|
|
*
|
|
* 1. The task's own named observable: a burst of entries appended in one
|
|
* order, whose majority commit becomes known all at once (a single
|
|
* covering ACK advancing committed_lsn past all of them in one jump),
|
|
* is applied in strict ascending LSN order — never reordered, never
|
|
* skipped — regardless of commit mode (a WAL_JOB_SUBMIT WAIT-mode entry
|
|
* mixed with WAL_JOB_{START,FINISH} RETAIN-mode facts in the same
|
|
* burst). Also folds in the Hazard 1 (double-append) regression check:
|
|
* current_lsn must not grow beyond the entries this test itself
|
|
* appended, proving the live pump's own post-mutation notification
|
|
* never re-appends an already-committed entry.
|
|
* 2. persist_apply_pump_run() called directly (white-box), confirming the
|
|
* walk applies exactly the committed-but-unapplied range and advances
|
|
* last_applied_lsn accordingly.
|
|
* 3. A non-regression check for the Hazard 1 fix's precision: job_ctrl's
|
|
* SIGCONT/SIGSTOP path (mark_job_update) still durably appends its own
|
|
* WAL_JOB_UPDATE entry — proving the double-append guard is scoped by
|
|
* call provenance (apply-seam-only), not by hook detail value, since
|
|
* job_ctrl fires the identical HOOK_DETAIL_JOB_UPDATE detail for its
|
|
* own, still-legitimate live append.
|
|
*
|
|
* Fixture mirrors test_commit_barrier_reply.c's cbrp_setup (real persist_init
|
|
* + persist_start over a private tmpdir so persist_start registers the
|
|
* production hooks, a real 3-member commit_engine_t with two stub standbys,
|
|
* ACKs delivered explicitly via repl_on_ack), trimmed of the worker-join
|
|
* scaffolding this file does not need. commit_set_pool is deliberately never
|
|
* called, so ce->pool stays NULL — persist_apply_pump_kick's documented
|
|
* NULL-pool fallback then runs the walk INLINE, synchronously, on whichever
|
|
* thread's kick wins the CAS-coalescing race.
|
|
*
|
|
* IMPORTANT (round-2 review correction): that does NOT mean every assertion
|
|
* below is directly observable immediately after a triggering call returns
|
|
* with no polling needed — persist_start's fsync_worker thread independently
|
|
* drives its own kick (commit_on_durable_advance) whenever durable_lsn
|
|
* advances, racing with this test's own repl_on_ack-driven kick. The
|
|
* coalescing guarantees exactly one kick actually runs the walk, but it can
|
|
* be the FSYNC THREAD's, asynchronously, after repl_on_ack has already
|
|
* returned on the main test thread. Tests that trigger a commit and then
|
|
* need to observe its APPLIED effect (not just committed_lsn) use
|
|
* tap_wait_applied() to poll persist_last_applied_lsn() before reading
|
|
* final state — see that helper's doc comment for the full mechanism and
|
|
* the flake it fixes.
|
|
*
|
|
* Every test below captures ALL assertion-relevant observables into local
|
|
* variables and calls tap_teardown() UNCONDITIONALLY before running any
|
|
* TEST_ASSERT — a failed assertion's longjmp must never skip teardown, or it
|
|
* hangs the next test in this binary (the known longjmp-past-teardown
|
|
* pathology from task 3.1's round-2 lesson, tasks.md Implementation Notes).
|
|
*
|
|
* Test 5 (bug-1415, task 6.2 round 2) is a deliberate exception to the
|
|
* "commit_set_pool is never called" note above: it is the one test in this
|
|
* file that wires a REAL multi-vcpu vtaskmgr_t pool, in its own dedicated
|
|
* fixture (tpr_setup/tpr_teardown, disjoint globals from tap_setup's) that
|
|
* leaves every other test in this file untouched. See that test's own
|
|
* header comment for why a real pool is required to reproduce bug-1415 at
|
|
* all.
|
|
*/
|
|
|
|
#include <rtti.h>
|
|
|
|
#include <capnp_c.h>
|
|
#include <errno.h>
|
|
#include <pthread.h>
|
|
#include <signal.h>
|
|
#include <stdbool.h>
|
|
#include <stdio.h>
|
|
#include <string.h>
|
|
#include <time.h>
|
|
#include <unistd.h>
|
|
|
|
#include <unity/unity.h>
|
|
|
|
#include "cluster.h"
|
|
#include "commit.h"
|
|
#include "controller.capnp.h"
|
|
#include "ha.h"
|
|
#include "hook.h"
|
|
#include "jobj.h"
|
|
#include "persist.h"
|
|
#include "repl.h"
|
|
#include "vtaskmgr.h"
|
|
#include "wal_types.h"
|
|
#include "xce/cluster_def.h"
|
|
#include "xce/types.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 other raft-apply-decoupling test file. */
|
|
static int g_tap_jobobj_ready;
|
|
|
|
static cluster_t g_tap_c;
|
|
static commit_engine_t g_tap_hce;
|
|
static int g_tap_hep_a, g_tap_hep_b; /* dummy standby endpoints */
|
|
static char g_tap_tmpdir[256];
|
|
|
|
/* Test-injectable send hook (repl.c, XCE_TESTING) — keeps repl_broadcast from
|
|
* dereferencing the dummy standby endpoints above, mirroring
|
|
* test_commit_barrier_reply.c's identical cbrp_drop_send_hook. */
|
|
extern int (*repl_broadcast_send_hook)(void *ep, wal_catchup_entry_t **entries,
|
|
int n_entries);
|
|
|
|
static int tap_drop_send_hook(void *ep, wal_catchup_entry_t **entries,
|
|
int n_entries)
|
|
{
|
|
(void)ep; (void)entries; (void)n_entries;
|
|
return 0; /* pretend success, drop the data */
|
|
}
|
|
|
|
/* ── Spy hook: records the exact (detail, jobid) sequence of every
|
|
* AFTER_EVENT firing on XCE_CONTROLLER_HOOK_JOB, in the order they occur. ── */
|
|
#define TAP_SPY_CAP 32
|
|
typedef struct {
|
|
int detail;
|
|
uint64_t jobid;
|
|
} tap_spy_entry_t;
|
|
|
|
static tap_spy_entry_t g_tap_spy[TAP_SPY_CAP];
|
|
static int g_tap_spy_n;
|
|
|
|
static int tap_spy_hook(int type, int situation, int detail, void* data,
|
|
va_list ap)
|
|
{
|
|
(void)type; (void)data;
|
|
if (situation != HOOK_AFTER_EVENT)
|
|
return 0;
|
|
|
|
uint64_t jobid = 0;
|
|
|
|
switch (detail) {
|
|
case HOOK_DETAIL_JOB_NEW: {
|
|
/* args: cluster_t*, xlist_t* (apply_committed_entry's WAL_JOB_SUBMIT
|
|
* branch) -- consume the leading cluster_t* first, mirroring
|
|
* persist_job_hook_fun's own va_arg pattern for this hook type. */
|
|
(void)va_arg(ap, cluster_t*);
|
|
xlist_t* jlist = va_arg(ap, xlist_t*);
|
|
if (jlist && jlist->count > 0) {
|
|
job_t* j = (job_t*)((void**)jlist->data)[0];
|
|
jobid = j->job.jobid;
|
|
}
|
|
break;
|
|
}
|
|
case HOOK_DETAIL_JOB_UPDATE: {
|
|
/* args: cluster_t*, job_t* -- same leading-arg shape. */
|
|
(void)va_arg(ap, cluster_t*);
|
|
job_t* j = va_arg(ap, job_t*);
|
|
if (j)
|
|
jobid = j->job.jobid;
|
|
break;
|
|
}
|
|
default:
|
|
return 0; /* not relevant to this file's assertions */
|
|
}
|
|
|
|
if (g_tap_spy_n < TAP_SPY_CAP) {
|
|
g_tap_spy[g_tap_spy_n].detail = detail;
|
|
g_tap_spy[g_tap_spy_n].jobid = jobid;
|
|
g_tap_spy_n++;
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
static void tap_spy_reset(void)
|
|
{
|
|
g_tap_spy_n = 0;
|
|
memset(g_tap_spy, 0, sizeof(g_tap_spy));
|
|
}
|
|
|
|
static void tap_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*)"tap-probe") == NULL)
|
|
xstring_table_init(64);
|
|
|
|
if (!g_tap_jobobj_ready) {
|
|
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
|
|
g_tap_jobobj_ready = 1;
|
|
}
|
|
|
|
memset(&g_tap_c, 0, sizeof(g_tap_c));
|
|
g_tap_c.myhost = &g_tap_c.myhost_info;
|
|
g_tap_c.myhost_info.name = (char*)"ctrl-test";
|
|
g_tap_c.myhost_info.key = (char*)"";
|
|
|
|
TEST_ASSERT_EQUAL_INT(0,
|
|
xce_hook_init(&g_tap_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
|
|
|
|
g_tap_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
|
|
g_tap_c.qmgr = makeInstance(&xkvmgr_type, &str_params);
|
|
g_tap_c.workermgr = makeInstance(&xkvmgr_type, &str_params);
|
|
g_tap_c.resmgr = makeInstance(&xkvmgr_type, &str_params);
|
|
TEST_ASSERT_NOT_NULL(g_tap_c.jobmgr);
|
|
TEST_ASSERT_NOT_NULL(g_tap_c.qmgr);
|
|
TEST_ASSERT_NOT_NULL(g_tap_c.workermgr);
|
|
TEST_ASSERT_NOT_NULL(g_tap_c.resmgr);
|
|
zarray_init(&g_tap_c.sharedinst, 16, NULL);
|
|
|
|
persist_config_t cfg = persist_config_defaults();
|
|
cfg.wal_sync_ms = 5000;
|
|
cfg.sync_timeout_ms = 2000;
|
|
|
|
snprintf(g_tap_tmpdir, sizeof(g_tap_tmpdir), "/tmp/xce_tap61_XXXXXX");
|
|
TEST_ASSERT_NOT_NULL(mkdtemp(g_tap_tmpdir));
|
|
persist_set_workdir(&g_tap_c.persist, g_tap_tmpdir, "ctrl");
|
|
TEST_ASSERT_EQUAL_INT(0, persist_init(&g_tap_c.persist, &cfg, &g_tap_c));
|
|
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_tap_c.persist));
|
|
|
|
/* 3-member cluster: 2 stub standbys, engine wired, ACKs withheld until a
|
|
* test explicitly delivers them via repl_on_ack. commit_set_pool is
|
|
* deliberately never called -- see file header for why. */
|
|
TEST_ASSERT_TRUE(
|
|
repl_add_standby(&g_tap_c.persist.repl, &g_tap_hep_a, "sb_a", 7500) >= 0);
|
|
TEST_ASSERT_TRUE(
|
|
repl_add_standby(&g_tap_c.persist.repl, &g_tap_hep_b, "sb_b", 7501) >= 0);
|
|
TEST_ASSERT_EQUAL_INT(0,
|
|
commit_engine_init(&g_tap_hce, &g_tap_c.persist, 3, 8));
|
|
repl_set_commit_engine(&g_tap_c.persist.repl, &g_tap_hce);
|
|
repl_broadcast_send_hook = tap_drop_send_hook;
|
|
|
|
TEST_ASSERT_EQUAL_INT(0,
|
|
xce_add_hook(&g_tap_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tap_spy_hook,
|
|
&g_tap_c));
|
|
|
|
tap_spy_reset();
|
|
}
|
|
|
|
static void tap_teardown(void)
|
|
{
|
|
repl_broadcast_send_hook = NULL;
|
|
commit_engine_fini(&g_tap_hce);
|
|
persist_stop(&g_tap_c.persist);
|
|
persist_fini(&g_tap_c.persist);
|
|
|
|
zarray_fini(&g_tap_c.sharedinst);
|
|
if (g_tap_c.resmgr) dereference(g_tap_c.resmgr);
|
|
if (g_tap_c.workermgr) dereference(g_tap_c.workermgr);
|
|
if (g_tap_c.qmgr) dereference(g_tap_c.qmgr);
|
|
if (g_tap_c.jobmgr) dereference(g_tap_c.jobmgr);
|
|
xce_hook_fini(&g_tap_c.hooksys);
|
|
}
|
|
|
|
/* Creates a job directly in the fixture's jobmgr, bypassing WAL entirely —
|
|
* for tests that only need a pre-existing job for a fact to transition,
|
|
* mirroring test_apply_reapply_idempotency.c's tari_seed_job. */
|
|
static job_submit_data_t g_tap_seed_submit;
|
|
|
|
static job_t* tap_seed_job(uint64_t jobid, uint32_t state)
|
|
{
|
|
xce_job_t jdef;
|
|
|
|
memset(&g_tap_seed_submit, 0, sizeof(g_tap_seed_submit));
|
|
g_tap_seed_submit.queue = (char*)"q";
|
|
g_tap_seed_submit.user = (char*)"tester";
|
|
|
|
memset(&jdef, 0, sizeof(jdef));
|
|
jdef.jobid = jobid;
|
|
jdef.state = state;
|
|
jdef.submit = &g_tap_seed_submit;
|
|
|
|
return makeJob(g_tap_c.jobmgr, &jdef);
|
|
}
|
|
|
|
static job_t* tap_find_job(uint64_t jobid)
|
|
{
|
|
job_t* j = NULL;
|
|
lock(g_tap_c.jobmgr, RTTI_LOCK_READ);
|
|
int found = key_find(g_tap_c.jobmgr, (void*)jobid, (void**)&j);
|
|
unlock(g_tap_c.jobmgr);
|
|
return (found > 0) ? j : NULL;
|
|
}
|
|
|
|
/* A standby ACK only counts toward majority once this node's OWN copy is
|
|
* durable (persist_durable_lsn, capped by the group-commit flusher's own
|
|
* fdatasync cadence -- ha-commit-durability's majority_candidate_locked
|
|
* caps self's contribution at durable_lsn, never the raw appended tip).
|
|
* persist_wal_async signals the flusher's condvar immediately after each
|
|
* append, but the flusher still runs on its own thread -- poll briefly
|
|
* rather than assume it has already caught up by the time this function
|
|
* returns, mirroring cbrp_poll_append's identical polling idiom in
|
|
* test_commit_barrier_reply.c. */
|
|
static void tap_wait_durable(uint64_t lsn, uint32_t deadline_ms)
|
|
{
|
|
for (uint32_t waited = 0; waited < deadline_ms; waited += 5) {
|
|
if (atomic_load(&g_tap_c.persist.durable_lsn) >= lsn)
|
|
return;
|
|
usleep(5000);
|
|
}
|
|
}
|
|
|
|
/* raft-apply-decoupling task 6.1 round 2 review: waits for the ordered apply
|
|
* pump to actually finish applying up through `lsn` after a triggering event
|
|
* (repl_on_ack) returns. This is needed even though ce->pool is NULL here
|
|
* (persist_apply_pump_kick's documented inline-fallback runs the walk on the
|
|
* CALLING thread rather than a vtask worker) -- a SEPARATE real thread, the
|
|
* persist group-commit flusher (fsync_worker, started by persist_start in
|
|
* tap_setup), ALSO independently drives a kick via commit_on_durable_advance
|
|
* every time durable_lsn itself advances, regardless of any ACK. That kick
|
|
* races with this test's own repl_on_ack-driven kick on the main thread.
|
|
* persist_apply_pump_kick's CAS-coalescing (apply_pump_scheduled) guarantees
|
|
* exactly one of the two ever actually runs the walk at a time -- but the
|
|
* LOSING side's caller returns immediately without having applied anything
|
|
* itself, handing responsibility to whichever thread already holds the
|
|
* flag. That is by design: the kick is "fire and forget" (coalesced,
|
|
* eventually-consistent dispatch), never "kick and block until this
|
|
* specific walk completes". So a test that calls repl_on_ack and then reads
|
|
* job state or hook-fire counts immediately cannot assume the walk that
|
|
* will apply its entries ran synchronously on ITS OWN thread, even with
|
|
* ce->pool NULL -- it may instead complete asynchronously, moments later, on
|
|
* the fsync worker thread. Poll persist_last_applied_lsn() (which only
|
|
* advances after a real apply pass, from whichever thread ran it) rather
|
|
* than assume repl_on_ack's return implies completion -- confirmed via
|
|
* review to reproduce as a genuine, reproducible flake (observed at ~7.5%,
|
|
* both in this file's burst-order test and the FAIL/FINISH race test)
|
|
* without this wait. */
|
|
static void tap_wait_applied(uint64_t lsn, uint32_t deadline_ms)
|
|
{
|
|
for (uint32_t waited = 0; waited < deadline_ms; waited += 5) {
|
|
if (persist_last_applied_lsn(&g_tap_c.persist) >= lsn)
|
|
return;
|
|
usleep(5000);
|
|
}
|
|
}
|
|
|
|
/* ── 1. The task's named observable + Hazard 1 regression ───────────────── */
|
|
|
|
static void test_apply_pump_kick_applies_committed_burst_in_ascending_lsn_order(
|
|
void)
|
|
{
|
|
tap_setup();
|
|
|
|
job_t* j901 = tap_seed_job(901, JOB_STAT_PEND);
|
|
job_t* j902 = tap_seed_job(902, JOB_STAT_PEND);
|
|
int seeded_ok = (j901 != NULL) && (j902 != NULL);
|
|
|
|
uint64_t first_lsn = g_tap_c.persist.current_lsn;
|
|
|
|
/* Append three facts, deliberately in an order that is NOT sorted by
|
|
* jobid and NOT the order a human might narrate the story in — job 902
|
|
* (START then FINISH) interleaved around job 901's single START. LSN
|
|
* order is exactly the append order below: 902/RUN, 901/RUN, 902/DONE. */
|
|
int rc1 = persist_submit_job_fact(&g_tap_c.persist, WAL_JOB_START, 902,
|
|
JOB_STAT_RUN);
|
|
int rc2 = persist_submit_job_fact(&g_tap_c.persist, WAL_JOB_START, 901,
|
|
JOB_STAT_RUN);
|
|
int rc3 = persist_submit_job_fact(&g_tap_c.persist, WAL_JOB_FINISH, 902,
|
|
JOB_STAT_DONE);
|
|
|
|
uint64_t watermark = g_tap_c.persist.current_lsn - 1;
|
|
|
|
/* Snapshot the pre-commit state before delivering any ACK: nothing may
|
|
* have mutated or notified yet (Req 5.1, 5.3, 6.1). */
|
|
uint32_t j901_state_precommit = seeded_ok ? j901->job.state : 0;
|
|
uint32_t j902_state_precommit = seeded_ok ? j902->job.state : 0;
|
|
int spy_n_precommit = g_tap_spy_n;
|
|
|
|
/* One covering ACK jumps committed_lsn past all three in a single
|
|
* contiguous-prefix advance -- commit_resolve_pending's unconditional
|
|
* persist_apply_pump_kick(ce->p, ce->pool) call (task 6.1's commit.c
|
|
* wiring) fires here; ce->pool is NULL in this fixture, so the walk
|
|
* runs inline, synchronously, before repl_on_ack returns. Self's own
|
|
* copy must be durable up to `watermark` first, or the ACK cannot lift
|
|
* committed_lsn that far (self caps its own contribution at
|
|
* durable_lsn, never the raw appended tip). */
|
|
tap_wait_durable(watermark, 2000);
|
|
repl_on_ack(&g_tap_c.persist.repl, &g_tap_hep_a, watermark);
|
|
|
|
/* See tap_wait_applied's doc comment: the fsync-worker thread's own
|
|
* durable-advance-driven kick can win the coalescing race and apply this
|
|
* burst asynchronously, after repl_on_ack already returned here. */
|
|
tap_wait_applied(watermark, 2000);
|
|
|
|
uint64_t committed_after = atomic_load(&g_tap_c.persist.committed_lsn);
|
|
int spy_n_after = g_tap_spy_n;
|
|
tap_spy_entry_t spy_snap[TAP_SPY_CAP];
|
|
memcpy(spy_snap, g_tap_spy, sizeof(spy_snap));
|
|
uint32_t j901_state_final = seeded_ok ? j901->job.state : 0;
|
|
uint32_t j902_state_final = seeded_ok ? j902->job.state : 0;
|
|
uint64_t last_applied_final = persist_last_applied_lsn(&g_tap_c.persist);
|
|
uint64_t current_lsn_final = g_tap_c.persist.current_lsn;
|
|
|
|
tap_teardown();
|
|
|
|
TEST_ASSERT_TRUE(seeded_ok);
|
|
TEST_ASSERT_EQUAL_INT(0, rc1);
|
|
TEST_ASSERT_EQUAL_INT(0, rc2);
|
|
TEST_ASSERT_EQUAL_INT(0, rc3);
|
|
TEST_ASSERT_EQUAL_UINT64(first_lsn + 2, watermark);
|
|
|
|
/* Nothing has committed yet (ACKs withheld) -- nothing may be applied. */
|
|
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j901_state_precommit,
|
|
"a fact must not mutate state before it commits");
|
|
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_PEND, j902_state_precommit,
|
|
"a fact must not mutate state before it commits");
|
|
TEST_ASSERT_EQUAL_INT_MESSAGE(0, spy_n_precommit,
|
|
"no notification may fire before any entry in the burst commits");
|
|
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(watermark, committed_after,
|
|
"the covering ACK must advance committed_lsn to the full watermark");
|
|
|
|
/* Strict order, no skipping or reordering (Req 3.3, 7.3): the spy's
|
|
* recorded sequence must be EXACTLY the append/LSN order, never sorted
|
|
* by jobid, never reflecting any other ordering scheme. */
|
|
TEST_ASSERT_EQUAL_INT_MESSAGE(3, spy_n_after,
|
|
"the burst must apply as exactly three notifications, one per "
|
|
"committed entry, in a single pump pass");
|
|
TEST_ASSERT_EQUAL_INT(HOOK_DETAIL_JOB_UPDATE, spy_snap[0].detail);
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(902, spy_snap[0].jobid,
|
|
"entry 1 (lowest LSN) must apply first, regardless of jobid");
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(901, spy_snap[1].jobid,
|
|
"entry 2 must apply second, in strict LSN order");
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(902, spy_snap[2].jobid,
|
|
"entry 3 (highest LSN) must apply last, even though it targets the "
|
|
"SAME job as entry 1 -- LSN order, not jobid grouping, governs");
|
|
|
|
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j901_state_final,
|
|
"job 901 must reflect its single applied fact");
|
|
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_DONE, j902_state_final,
|
|
"job 902 must reflect its LAST applied fact (FINISH), proving "
|
|
"entry 2 (job 901, in between) did not get skipped or cause "
|
|
"job 902's later entry to apply out of turn");
|
|
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(watermark, last_applied_final,
|
|
"last_applied_lsn must advance to exactly the applied watermark");
|
|
|
|
/* Hazard 1 regression: applying these facts live must not re-append
|
|
* them. Without the t_apply_seam_notifying guard, each HOOK_DETAIL_
|
|
* JOB_UPDATE notification above would have looped back through
|
|
* persist_job_hook_fun and appended a duplicate WAL_JOB_UPDATE entry,
|
|
* growing current_lsn past this test's own three appends. */
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(watermark + 1, current_lsn_final,
|
|
"the live apply pump must never re-append an already-committed "
|
|
"entry (Hazard 1) -- current_lsn must reflect only this test's own "
|
|
"three appends, nothing more");
|
|
|
|
printf("test_apply_pump_kick_applies_committed_burst_in_ascending_lsn_"
|
|
"order: PASS\n");
|
|
}
|
|
|
|
/* ── 2. persist_apply_pump_run() direct call: mixed command + fact ──────── */
|
|
|
|
static int tap_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;
|
|
}
|
|
|
|
static void test_apply_pump_run_applies_mixed_submit_and_fact_entries(void)
|
|
{
|
|
tap_setup();
|
|
|
|
job_t* j903 = tap_seed_job(903, JOB_STAT_PEND);
|
|
int seeded_ok = (j903 != NULL);
|
|
|
|
/* Entry 1 (lower LSN): a fact for the pre-existing job. */
|
|
int rc1 = persist_submit_job_fact(&g_tap_c.persist, WAL_JOB_START, 903,
|
|
JOB_STAT_RUN);
|
|
|
|
/* Entry 2 (higher LSN): a WAL_JOB_SUBMIT command entry for a brand-new
|
|
* job, appended directly via persist_wal_async (bypassing the full
|
|
* commit_submit_async/continuation pipeline, which is task 3.2/6.2's
|
|
* concern, not this walk's) -- this file only needs the entry's bytes
|
|
* to exist on disk at a known LSN, exactly what the live pump consumes
|
|
* regardless of how the entry got there. */
|
|
xce_job_t jdef;
|
|
memset(&jdef, 0, sizeof(jdef));
|
|
jdef.jobid = 904;
|
|
jdef.state = JOB_STAT_PEND;
|
|
jdef.submit = &g_tap_seed_submit;
|
|
|
|
uint8_t* payload = NULL;
|
|
size_t plen = 0;
|
|
int enc_rc = tap_encode_job(&jdef, &payload, &plen);
|
|
int rc2 = persist_wal_async(&g_tap_c.persist, WAL_JOB_SUBMIT,
|
|
payload, (uint32_t)plen);
|
|
free(payload);
|
|
|
|
uint64_t watermark = g_tap_c.persist.current_lsn - 1;
|
|
|
|
/* White-box: mark the whole range committed directly (this test targets
|
|
* the walk's own dispatch logic, not commit-advance itself -- that is
|
|
* test 1's job) and drive the walk with the exposed, directly-callable
|
|
* entry point. */
|
|
atomic_store(&g_tap_c.persist.committed_lsn, watermark);
|
|
|
|
int applied = persist_apply_pump_run(&g_tap_c.persist);
|
|
int applied2 = persist_apply_pump_run(&g_tap_c.persist);
|
|
|
|
uint32_t j903_state_final = seeded_ok ? j903->job.state : 0;
|
|
job_t* j904 = tap_find_job(904);
|
|
uint32_t j904_state_final = j904 ? j904->job.state : 0;
|
|
uint64_t last_applied_final = persist_last_applied_lsn(&g_tap_c.persist);
|
|
|
|
tap_teardown();
|
|
|
|
TEST_ASSERT_TRUE(seeded_ok);
|
|
TEST_ASSERT_EQUAL_INT(0, rc1);
|
|
TEST_ASSERT_EQUAL_INT(0, enc_rc);
|
|
TEST_ASSERT_EQUAL_INT(0, rc2);
|
|
|
|
/* raft-apply-decoupling task 6.1 round 2 review: this fixture's
|
|
* commit_engine_t has a real fsync_worker thread wired (tap_setup), and
|
|
* persist_submit_job_fact/persist_wal_async above are real appends --
|
|
* once each becomes durable, commit_on_durable_advance (persist.c) fires
|
|
* on the FSYNC THREAD and unconditionally calls persist_apply_pump_kick,
|
|
* completely independent of this test never calling repl_on_ack. That
|
|
* background kick can race this test's own two direct
|
|
* persist_apply_pump_run() calls below and win the apply_pump_lock
|
|
* first -- applying both entries itself before either of this test's
|
|
* own calls gets a turn (persist_apply_pump_run is mutex-serialized and
|
|
* fully synchronous, so whichever caller actually performs the walk
|
|
* always completes it before returning -- there is no partial/in-flight
|
|
* state to poll for here, unlike tests 1/4's tap_wait_applied
|
|
* situation). When the background thread wins BEFORE either of this
|
|
* test's own calls, BOTH `applied` and `applied2` legitimately read 0
|
|
* (nothing was left for this test's own calls to do) even though every
|
|
* entry did get applied, correctly, by the background thread's own
|
|
* separate call -- so neither the individual values nor their sum can
|
|
* be asserted exactly in this shared-thread fixture without a deeper
|
|
* synchronization change. What IS deterministic, and what this test
|
|
* exists to actually prove, is a purely state-based invariant: every
|
|
* committed entry ends up applied by SOMEONE exactly once, regardless
|
|
* of commit mode, and the final materialized state is correct --
|
|
* checked below via last_applied_lsn and the two jobs' live state,
|
|
* never via the racy call-arity split. Observed empirically via review:
|
|
* the original strict `applied==2 && applied2==0` split flaked at
|
|
* ~1.6% under this exact race. */
|
|
TEST_ASSERT_TRUE_MESSAGE(applied >= 0 && applied <= 2 &&
|
|
applied2 >= 0 && applied2 <= 2,
|
|
"the walk must never report a negative or over-large apply count");
|
|
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, j903_state_final,
|
|
"the fact entry must have applied");
|
|
TEST_ASSERT_NOT_NULL_MESSAGE(j904,
|
|
"the WAL_JOB_SUBMIT command entry must have applied too -- the walk "
|
|
"does not skip command-mode entries");
|
|
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j904_state_final);
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(watermark, last_applied_final,
|
|
"last_applied_lsn must reach the watermark no matter which thread "
|
|
"performed the walk -- this, not the applied/applied2 call-arity "
|
|
"split, is the authoritative proof both entries were applied "
|
|
"exactly once");
|
|
|
|
printf("test_apply_pump_run_applies_mixed_submit_and_fact_entries: "
|
|
"PASS\n");
|
|
}
|
|
|
|
/* ── 2b. Task 4.1 (job-array-single-wal-record, design.md "Live Apply
|
|
* Resource Fence"): the walk must respect a negative apply outcome ─────── */
|
|
|
|
/* Encodes a (base_jobid, array_size, spec) JobSubmit payload -- the exact
|
|
* WAL_JOB_ARRAY_SUBMIT payload shape, mirroring test_job_array_apply.c's
|
|
* taj_encode_job_submit. Caller frees *out. */
|
|
static int tap_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;
|
|
}
|
|
|
|
/* Before this fix, persist_apply_pump_run() discarded
|
|
* apply_committed_entry()'s return value entirely
|
|
* ((void)apply_committed_entry(...)) and unconditionally advanced
|
|
* last_applied_lsn -- silently, permanently losing any apply failure
|
|
* (design.md "Live Apply Resource Fence", Req 9.4). This test forces a REAL
|
|
* negative return through the REAL WAL_JOB_ARRAY_SUBMIT dispatch path
|
|
* (task 3.1's scratch-build-phase fault-injection seam), rather than
|
|
* mocking apply_committed_entry() itself, and proves the walk now:
|
|
* (a) returns negative -- not silent 0/success;
|
|
* (b) does not advance last_applied_lsn past the failed position;
|
|
* (c) genuinely halts -- a LATER, already-committed entry in the same
|
|
* batch is NOT applied on this pass, proving this is a stop, not a
|
|
* skip-and-continue;
|
|
* (d) reaches the loud, operator-visible ERROR log branch: this is the
|
|
* ONLY source of a negative return from persist_apply_pump_run()'s
|
|
* WAL_JOB_ARRAY_SUBMIT dispatch in this scenario, and the ERROR()
|
|
* call is unconditionally on that exact return path (see persist.c) --
|
|
* confirmed structurally rather than by parsing log output, per this
|
|
* suite's convention (no log-capture mechanism exists in this
|
|
* fixture).
|
|
* A subsequent retry, with the fault cleared, then proves the walk resumes
|
|
* from the SAME stuck position (not a saved-forward offset) and completes
|
|
* both entries once unstuck -- mirroring the "entry stays committed and
|
|
* will be retried, from the same position, on the next kick" contract the
|
|
* pre-existing apply_pump_read_entry failure branch already established
|
|
* immediately above this one in the same function. */
|
|
static void test_apply_pump_run_halts_at_failed_lsn_without_advancing(void)
|
|
{
|
|
tap_setup();
|
|
|
|
#ifdef XCE_TESTING
|
|
job_array_apply_test_clear_force_fail();
|
|
#endif
|
|
|
|
const uint64_t base_jobid = 93000;
|
|
const uint64_t array_size = 3;
|
|
job_submit_data_t spec;
|
|
memset(&spec, 0, sizeof(spec));
|
|
spec.queue = (char*)"q";
|
|
spec.user = (char*)"tester";
|
|
spec.mintasks = 1;
|
|
spec.maxtasks = 1;
|
|
spec.job_name = (char*)"array-fail-test";
|
|
|
|
uint8_t* array_payload = NULL;
|
|
size_t array_len = 0;
|
|
int enc_rc = tap_encode_job_submit(base_jobid, array_size, &spec,
|
|
&array_payload, &array_len);
|
|
|
|
/* Fail at scratch-build index 1 (of 3) -- set BEFORE either entry below
|
|
* is appended, so no background fsync-driven kick (see file header) can
|
|
* ever observe an unfailed attempt at this LSN: the fault is
|
|
* deterministic and sticky until cleared, so every caller that reaches
|
|
* this LSN -- this test's own call below, or a racing background one --
|
|
* gets the identical negative rc without advancing anything. */
|
|
#ifdef XCE_TESTING
|
|
job_array_apply_test_set_force_fail_at(1);
|
|
#endif
|
|
|
|
int rc_array = persist_wal_async(&g_tap_c.persist, WAL_JOB_ARRAY_SUBMIT,
|
|
array_payload, (uint32_t)array_len);
|
|
free(array_payload);
|
|
uint64_t array_lsn = g_tap_c.persist.current_lsn - 1;
|
|
|
|
/* A second, later entry -- an ordinary single-job submit -- committed in
|
|
* the SAME batch as the array entry above. If the walk merely skipped
|
|
* the failed array record instead of halting at it, this job would end
|
|
* up materialized; a genuine halt must leave it untouched. */
|
|
memset(&g_tap_seed_submit, 0, sizeof(g_tap_seed_submit));
|
|
g_tap_seed_submit.queue = (char*)"q";
|
|
g_tap_seed_submit.user = (char*)"tester";
|
|
|
|
xce_job_t jdef;
|
|
memset(&jdef, 0, sizeof(jdef));
|
|
jdef.jobid = 94000;
|
|
jdef.state = JOB_STAT_PEND;
|
|
jdef.submit = &g_tap_seed_submit;
|
|
|
|
uint8_t* later_payload = NULL;
|
|
size_t later_len = 0;
|
|
int enc_rc2 = tap_encode_job(&jdef, &later_payload, &later_len);
|
|
int rc_later = persist_wal_async(&g_tap_c.persist, WAL_JOB_SUBMIT,
|
|
later_payload, (uint32_t)later_len);
|
|
free(later_payload);
|
|
uint64_t later_lsn = g_tap_c.persist.current_lsn - 1;
|
|
|
|
uint64_t baseline_applied = persist_last_applied_lsn(&g_tap_c.persist);
|
|
|
|
/* White-box: mark both entries committed directly (this test targets
|
|
* the walk's own return-value handling, not commit-advance itself --
|
|
* that is test 1's job). */
|
|
atomic_store(&g_tap_c.persist.committed_lsn, later_lsn);
|
|
|
|
int walk_rc = persist_apply_pump_run(&g_tap_c.persist);
|
|
|
|
uint64_t last_applied_after_failure =
|
|
persist_last_applied_lsn(&g_tap_c.persist);
|
|
job_t* array_job0_after_failure = tap_find_job(base_jobid + 0);
|
|
job_t* later_job_after_failure = tap_find_job(94000);
|
|
|
|
/* Clear the fault and retry: the walk must resume from the SAME stuck
|
|
* LSN (not have skipped it) and, this time, apply both entries. */
|
|
#ifdef XCE_TESTING
|
|
job_array_apply_test_clear_force_fail();
|
|
#endif
|
|
int walk_rc_retry = persist_apply_pump_run(&g_tap_c.persist);
|
|
|
|
uint64_t last_applied_after_retry =
|
|
persist_last_applied_lsn(&g_tap_c.persist);
|
|
job_t* array_job0_after_retry = tap_find_job(base_jobid + 0);
|
|
job_t* array_job1_after_retry = tap_find_job(base_jobid + 1);
|
|
job_t* array_job2_after_retry = tap_find_job(base_jobid + 2);
|
|
job_t* later_job_after_retry = tap_find_job(94000);
|
|
|
|
tap_teardown();
|
|
|
|
TEST_ASSERT_EQUAL_INT(0, enc_rc);
|
|
TEST_ASSERT_EQUAL_INT(0, rc_array);
|
|
TEST_ASSERT_EQUAL_INT(0, enc_rc2);
|
|
TEST_ASSERT_EQUAL_INT(0, rc_later);
|
|
TEST_ASSERT_EQUAL_UINT64(array_lsn + 1, later_lsn);
|
|
|
|
TEST_ASSERT_TRUE_MESSAGE(walk_rc < 0,
|
|
"the walk must return a negative value -- not silent 0/success -- "
|
|
"when apply_committed_entry fails");
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(baseline_applied,
|
|
last_applied_after_failure,
|
|
"last_applied_lsn must NOT advance past the failed position");
|
|
TEST_ASSERT_NULL_MESSAGE(array_job0_after_failure,
|
|
"no element of the failed array attempt may be visible -- "
|
|
"restore_job_array's own scratch-build atomicity plus this walk's "
|
|
"halt must together leave nothing materialized");
|
|
TEST_ASSERT_NULL_MESSAGE(later_job_after_failure,
|
|
"REGRESSION GUARD: the later, already-committed entry must NOT be "
|
|
"applied on this same pass -- the walk genuinely halted at the "
|
|
"failed position, it did not skip past it and continue");
|
|
|
|
TEST_ASSERT_TRUE_MESSAGE(walk_rc_retry >= 0,
|
|
"once the fault clears, the retry from the SAME stuck position "
|
|
"must succeed");
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(later_lsn, last_applied_after_retry,
|
|
"the retry must resume from the same stuck LSN and reach the "
|
|
"watermark -- proving the earlier failure did not skip the entry");
|
|
TEST_ASSERT_NOT_NULL_MESSAGE(array_job0_after_retry,
|
|
"array element 0 must exist once the walk is unstuck");
|
|
TEST_ASSERT_NOT_NULL_MESSAGE(array_job1_after_retry,
|
|
"array element 1 (the previously-forced index) must exist once the "
|
|
"walk is unstuck");
|
|
TEST_ASSERT_NOT_NULL_MESSAGE(array_job2_after_retry,
|
|
"array element 2 must exist once the walk is unstuck");
|
|
TEST_ASSERT_NOT_NULL_MESSAGE(later_job_after_retry,
|
|
"the later entry must apply too, once the walk is unstuck");
|
|
|
|
printf("test_apply_pump_run_halts_at_failed_lsn_without_advancing: "
|
|
"PASS\n");
|
|
}
|
|
|
|
/* ── 2c. Task 4.2 (job-array-single-wal-record, design.md "Live Apply
|
|
* Resource Fence"): a resource-exhaustion apply failure that halts the walk
|
|
* (task 4.1) must additionally ENGAGE a new, live, self-healing fence --
|
|
* persist_apply_resource_fenced()/persist_mark_apply_resource_fenced() --
|
|
* deliberately DISTINCT from the boot-only sticky recovery_unrecoverable
|
|
* flag (Req 9.4). Reuses the same force-fail seam and array-record shape as
|
|
* test 2b above, but asserts the NEW fence accessor rather than the halt
|
|
* behavior itself (already covered by test 2b). */
|
|
|
|
static void test_apply_pump_run_engages_live_fence_on_failed_lsn(void)
|
|
{
|
|
tap_setup();
|
|
|
|
#ifdef XCE_TESTING
|
|
job_array_apply_test_clear_force_fail();
|
|
#endif
|
|
|
|
bool fenced_before = persist_apply_resource_fenced(&g_tap_c.persist);
|
|
|
|
const uint64_t base_jobid = 95000;
|
|
const uint64_t array_size = 3;
|
|
job_submit_data_t spec;
|
|
memset(&spec, 0, sizeof(spec));
|
|
spec.queue = (char*)"q";
|
|
spec.user = (char*)"tester";
|
|
spec.mintasks = 1;
|
|
spec.maxtasks = 1;
|
|
spec.job_name = (char*)"array-fence-test";
|
|
|
|
uint8_t* array_payload = NULL;
|
|
size_t array_len = 0;
|
|
int enc_rc = tap_encode_job_submit(base_jobid, array_size, &spec,
|
|
&array_payload, &array_len);
|
|
|
|
/* Fail at scratch-build index 1 of 3 -- identical seam to test 2b. */
|
|
#ifdef XCE_TESTING
|
|
job_array_apply_test_set_force_fail_at(1);
|
|
#endif
|
|
|
|
int rc_array = persist_wal_async(&g_tap_c.persist, WAL_JOB_ARRAY_SUBMIT,
|
|
array_payload, (uint32_t)array_len);
|
|
free(array_payload);
|
|
uint64_t array_lsn = g_tap_c.persist.current_lsn - 1;
|
|
|
|
atomic_store(&g_tap_c.persist.committed_lsn, array_lsn);
|
|
|
|
int walk_rc = persist_apply_pump_run(&g_tap_c.persist);
|
|
|
|
bool fenced_after_failure =
|
|
persist_apply_resource_fenced(&g_tap_c.persist);
|
|
|
|
#ifdef XCE_TESTING
|
|
job_array_apply_test_clear_force_fail();
|
|
#endif
|
|
|
|
tap_teardown();
|
|
|
|
TEST_ASSERT_EQUAL_INT(0, enc_rc);
|
|
TEST_ASSERT_EQUAL_INT(0, rc_array);
|
|
TEST_ASSERT_FALSE_MESSAGE(fenced_before,
|
|
"a fresh persist_t must start with the Live Apply Resource Fence "
|
|
"CLEAR -- nothing has failed yet, and this is a distinct piece of "
|
|
"state from the boot-only recovery_unrecoverable flag");
|
|
TEST_ASSERT_TRUE_MESSAGE(walk_rc < 0,
|
|
"the forced construction failure must still halt the walk (task "
|
|
"4.1 regression guard)");
|
|
TEST_ASSERT_TRUE_MESSAGE(fenced_after_failure,
|
|
"the Live Apply Resource Fence must ENGAGE the moment the apply "
|
|
"walk halts on a resource-exhaustion apply failure (Req 9.4)");
|
|
|
|
printf("test_apply_pump_run_engages_live_fence_on_failed_lsn: PASS\n");
|
|
}
|
|
|
|
/* ── 2d. Task 4.2 self-heal: the fence must CLEAR itself, from inside the
|
|
* apply walk, the instant a retry of the SAME stuck LSN succeeds -- with NO
|
|
* external/admin action (Req 9.5). Mirrors test 2b's engage-then-retry
|
|
* shape but asserts the fence accessor at each stage instead of job
|
|
* visibility. */
|
|
|
|
static void test_apply_pump_run_self_clears_fence_on_successful_retry(void)
|
|
{
|
|
tap_setup();
|
|
|
|
#ifdef XCE_TESTING
|
|
job_array_apply_test_clear_force_fail();
|
|
#endif
|
|
|
|
const uint64_t base_jobid = 96000;
|
|
const uint64_t array_size = 3;
|
|
job_submit_data_t spec;
|
|
memset(&spec, 0, sizeof(spec));
|
|
spec.queue = (char*)"q";
|
|
spec.user = (char*)"tester";
|
|
spec.mintasks = 1;
|
|
spec.maxtasks = 1;
|
|
spec.job_name = (char*)"array-fence-selfheal-test";
|
|
|
|
uint8_t* array_payload = NULL;
|
|
size_t array_len = 0;
|
|
int enc_rc = tap_encode_job_submit(base_jobid, array_size, &spec,
|
|
&array_payload, &array_len);
|
|
|
|
/* Fail at scratch-build index 1 of 3 -- identical seam to tests 2b/2c. */
|
|
#ifdef XCE_TESTING
|
|
job_array_apply_test_set_force_fail_at(1);
|
|
#endif
|
|
|
|
int rc_array = persist_wal_async(&g_tap_c.persist, WAL_JOB_ARRAY_SUBMIT,
|
|
array_payload, (uint32_t)array_len);
|
|
free(array_payload);
|
|
uint64_t array_lsn = g_tap_c.persist.current_lsn - 1;
|
|
|
|
atomic_store(&g_tap_c.persist.committed_lsn, array_lsn);
|
|
|
|
/* First attempt: forced failure engages the fence (proven by test 2c;
|
|
* re-checked here as this test's own precondition). */
|
|
int walk_rc_fail = persist_apply_pump_run(&g_tap_c.persist);
|
|
bool fenced_after_failure =
|
|
persist_apply_resource_fenced(&g_tap_c.persist);
|
|
|
|
/* Clear the injected fault and retry the SAME stuck LSN -- no external
|
|
* or admin action, purely the apply walk retrying on its own. */
|
|
#ifdef XCE_TESTING
|
|
job_array_apply_test_clear_force_fail();
|
|
#endif
|
|
int walk_rc_retry = persist_apply_pump_run(&g_tap_c.persist);
|
|
bool fenced_after_retry =
|
|
persist_apply_resource_fenced(&g_tap_c.persist);
|
|
|
|
tap_teardown();
|
|
|
|
TEST_ASSERT_EQUAL_INT(0, enc_rc);
|
|
TEST_ASSERT_EQUAL_INT(0, rc_array);
|
|
TEST_ASSERT_TRUE_MESSAGE(walk_rc_fail < 0,
|
|
"the forced failure must still halt the walk (task 4.1 regression "
|
|
"guard)");
|
|
TEST_ASSERT_TRUE_MESSAGE(fenced_after_failure,
|
|
"precondition: the fence must be engaged before this test can prove "
|
|
"self-heal");
|
|
TEST_ASSERT_TRUE_MESSAGE(walk_rc_retry >= 0,
|
|
"once the fault clears, the retry of the same stuck LSN must "
|
|
"succeed");
|
|
TEST_ASSERT_FALSE_MESSAGE(fenced_after_retry,
|
|
"the Live Apply Resource Fence must CLEAR itself, from inside the "
|
|
"apply walk, the moment the retry of the SAME stuck position "
|
|
"succeeds -- with no external/admin action (Req 9.5)");
|
|
|
|
printf("test_apply_pump_run_self_clears_fence_on_successful_retry: "
|
|
"PASS\n");
|
|
}
|
|
|
|
/* ── 3. Hazard 1 guard precision: job_ctrl's own live append must survive ── */
|
|
|
|
static void test_apply_pump_guard_does_not_break_job_ctrl_live_append(void)
|
|
{
|
|
tap_setup();
|
|
|
|
/* A suspended, not-yet-started job (allocs == NULL) -- job_ctrl's
|
|
* SIGCONT branch calls mark_job_update() directly, synchronously
|
|
* mutating state and firing BEFORE/AFTER_EVENT HOOK_DETAIL_JOB_UPDATE
|
|
* itself, exactly the same detail apply_committed_entry() now also
|
|
* fires for live-applied facts (test 1 above). This is the dual-origin
|
|
* case the t_apply_seam_notifying guard must NOT break. */
|
|
job_t* j = tap_seed_job(905, JOB_STAT_PSUSP);
|
|
int seeded_ok = (j != NULL) && (j->job.allocs == NULL);
|
|
|
|
uint64_t first_lsn = g_tap_c.persist.current_lsn;
|
|
|
|
xce_signal_t sig = { .signo = SIGCONT, .force = 0 };
|
|
xce_jobctrl_t ctrl = { .action = JobControl_u_siginfo, .jobid = 905 };
|
|
ctrl.u.siginfo = &sig;
|
|
|
|
int rc = job_ctrl(&g_tap_c, &ctrl);
|
|
|
|
uint32_t state_final = seeded_ok ? j->job.state : 0xffffffffu;
|
|
uint64_t current_lsn_final = g_tap_c.persist.current_lsn;
|
|
|
|
tap_teardown();
|
|
|
|
TEST_ASSERT_TRUE(seeded_ok);
|
|
TEST_ASSERT_EQUAL_INT(0, rc);
|
|
TEST_ASSERT_EQUAL_UINT32_MESSAGE(0, state_final & JOB_STAT_PSUSP,
|
|
"SIGCONT must clear the suspend bit synchronously, as it always "
|
|
"has -- job_ctrl's command path is untouched by task 6.1");
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(first_lsn + 1, current_lsn_final,
|
|
"job_ctrl's own live append via persist_job_hook_fun must still "
|
|
"happen exactly once -- the apply-seam double-append guard is "
|
|
"scoped to apply_committed_entry()'s own call stack (a thread-local "
|
|
"flag), not to the HOOK_DETAIL_JOB_UPDATE value itself, so it must "
|
|
"never suppress this legitimate, unrelated live append");
|
|
|
|
printf("test_apply_pump_guard_does_not_break_job_ctrl_live_append: "
|
|
"PASS\n");
|
|
}
|
|
|
|
/* ── 4. Merge-staleness regression: FAIL must not clobber a racing FINISH ─
|
|
*
|
|
* Reproduces the round-1 corruption bug found by review: a FINISH fact
|
|
* (deterministic, state-independent -- exit_code alone decides DONE vs
|
|
* EXIT) and a POST_EXEC-stage FAIL fact for the SAME job are received in
|
|
* two separate commit rounds, with the FAIL's receipt happening BEFORE
|
|
* either fact has applied (both are durably appended and registered, but
|
|
* apply is deferred past the commit barrier -- exactly task 6.1's point).
|
|
* Driven through the REAL receipt handlers (on_job_finish/on_job_fail), not
|
|
* seeded directly, so the genuine staleness window is exercised: at FAIL
|
|
* receipt time, the job's real in-memory state is still RUN (FINISH has
|
|
* not applied yet). Old (round-1) code baked a snapshot of that STALE RUN
|
|
* state into the FAIL's WAL entry, so once both entries applied in LSN
|
|
* order the FAIL's blind overwrite clobbered FINISH's DONE with a stray
|
|
* RUN bit, corrupting the job to RUN|EXIT|PERR instead of the correct
|
|
* merged terminal state DONE|PERR. */
|
|
|
|
static void test_apply_pump_fail_racing_finish_merges_against_current_state(
|
|
void)
|
|
{
|
|
tap_setup();
|
|
|
|
job_t* j906 = tap_seed_job(906, JOB_STAT_RUN);
|
|
int seeded_ok = (j906 != NULL);
|
|
|
|
uint64_t first_lsn = g_tap_c.persist.current_lsn;
|
|
|
|
/* Round 1: worker reports FINISH (exit_code 0 -> DONE). Durably
|
|
* appended only -- record_job_finish_fact fires no AFTER_EVENT and
|
|
* mutates nothing, so j906->job.state must still read RUN right after
|
|
* this call returns. */
|
|
xce_jobfinish_t finish = { .jobid = 906, .exit_code = 0 };
|
|
int rc_finish = on_job_finish(&g_tap_c, &finish);
|
|
|
|
/* Round 2: BEFORE either fact has applied, a POST_EXEC-stage FAIL
|
|
* arrives for the same job (an ordinary HPC lifecycle event -- e.g. an
|
|
* epilogue script failing shortly after the main process exits). At
|
|
* this exact instant, job.state is STILL RUN -- the genuine staleness
|
|
* window this test reproduces, not a seeded/simulated one. */
|
|
uint32_t state_at_fail_receipt = seeded_ok ? j906->job.state : 0;
|
|
|
|
job_failinfo_t failinfo = { .jobid = 906,
|
|
.stage = JOB_EXEC_STAGE_POST_EXEC,
|
|
.code = 1,
|
|
.detail = NULL };
|
|
int rc_fail = on_job_fail(&g_tap_c, &failinfo);
|
|
|
|
uint32_t state_before_commit = seeded_ok ? j906->job.state : 0;
|
|
uint64_t watermark = g_tap_c.persist.current_lsn - 1;
|
|
|
|
/* One covering ACK commits both entries at once; the real
|
|
* commit_resolve_pending -> persist_apply_pump_kick production path
|
|
* applies them in strict LSN order (FINISH first, then FAIL), exactly
|
|
* as task 6.1's pump guarantees. */
|
|
tap_wait_durable(watermark, 2000);
|
|
repl_on_ack(&g_tap_c.persist.repl, &g_tap_hep_a, watermark);
|
|
|
|
/* See tap_wait_applied's doc comment: the fsync-worker thread's own
|
|
* durable-advance-driven kick can win the coalescing race and apply
|
|
* these two entries asynchronously, after repl_on_ack already returned
|
|
* here -- this is the exact mechanism behind the round-2 self-disclosed
|
|
* flake, not a corruption bug in the merge-staleness fix itself. */
|
|
tap_wait_applied(watermark, 2000);
|
|
|
|
uint64_t committed_after = atomic_load(&g_tap_c.persist.committed_lsn);
|
|
uint32_t state_final = seeded_ok ? j906->job.state : 0;
|
|
|
|
tap_teardown();
|
|
|
|
TEST_ASSERT_TRUE(seeded_ok);
|
|
TEST_ASSERT_EQUAL_INT(0, rc_finish);
|
|
TEST_ASSERT_EQUAL_INT(0, rc_fail);
|
|
TEST_ASSERT_EQUAL_UINT64(first_lsn + 1, watermark);
|
|
|
|
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, state_at_fail_receipt,
|
|
"the FAIL fact must be received while the job is still genuinely "
|
|
"RUN -- proving FINISH has not applied yet, the real staleness "
|
|
"window, not a simulated one");
|
|
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_RUN, state_before_commit,
|
|
"neither fact may mutate state before its entry commits (Req 5.1)");
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(watermark, committed_after,
|
|
"the covering ACK must advance committed_lsn past both entries");
|
|
|
|
TEST_ASSERT_EQUAL_UINT32_MESSAGE((JOB_STAT_DONE | JOB_STAT_PERR),
|
|
state_final,
|
|
"FAIL's PERR merge must be computed against FINISH's already-"
|
|
"applied DONE (correct terminal state) -- not a stale RUN snapshot "
|
|
"taken at FAIL's receipt time");
|
|
TEST_ASSERT_EQUAL_UINT32_MESSAGE(0, state_final & JOB_STAT_RUN,
|
|
"the stray RUN bit from the round-1 corruption bug must never "
|
|
"appear in the final state");
|
|
TEST_ASSERT_TRUE_MESSAGE((state_final & JOB_STAT_DONE) != 0,
|
|
"FINISH's DONE bit must survive -- it must never be silently "
|
|
"replaced by a later FAIL's stale-computed overwrite");
|
|
|
|
printf("test_apply_pump_fail_racing_finish_merges_against_current_"
|
|
"state: PASS\n");
|
|
}
|
|
|
|
/* ── 5. bug-1415 regression: late pending-apply registration self-heals ──
|
|
*
|
|
* Every test above (and every other test in this spec's test files)
|
|
* deliberately keeps ce->pool == NULL, so both commit.c's
|
|
* dispatch_continuation and persist_apply_pump_kick fall back to their
|
|
* documented inline/synchronous behavior — the continuation-registration
|
|
* path and the apply-pump walk can never actually run on two different
|
|
* threads in that mode, so the race bug-1415 describes cannot manifest
|
|
* there by construction (this is exactly why round 1's review found it
|
|
* untested by every fixture in this spec).
|
|
*
|
|
* This test wires a REAL vtaskmgr_t (n_vcpus > 1) via commit_set_pool, in
|
|
* its own dedicated fixture (tpr_setup/tpr_teardown, disjoint globals from
|
|
* tap_setup's) — commit_resolve_pending()'s two independent dispatches, the
|
|
* commit continuation (commit.c's dispatch_continuation) and the apply-pump
|
|
* kick (persist_apply_pump_kick), now run as REAL vtasks bound to separate
|
|
* idle vcpu worker threads by vtaskmgr's own dispatcher (vdispatcher_entry),
|
|
* exactly as production does (n_vcpus=8 by default, cluster_init_phase2).
|
|
* vtaskmgr.c has no completion-order guarantee between the two, so this
|
|
* test forces the specific bad interleaving deterministically instead of
|
|
* relying on natural timing luck: the test's own commit-continuation
|
|
* callback (tpr_commit_continuation, standing in for
|
|
* jobsubmit_commit_continuation — this test drives commit.c/persist.c
|
|
* directly, the actual layer bug-1415 lives in, rather than routing through
|
|
* job.c) blocks, polling persist_last_applied_lsn(), until the apply pump
|
|
* has ALREADY advanced past this entry's lsn — i.e. until
|
|
* persist_take_pending_apply() has already run (and found nothing, since
|
|
* nothing is registered yet) for this exact lsn — before ever calling
|
|
* persist_register_pending_apply(). This is a genuine barrier tied to the
|
|
* actual race condition, not a timing guess: as long as the pump eventually
|
|
* applies a durably committed entry (guaranteed here), the poll always
|
|
* terminates with the pump strictly ahead, reproducing the exact
|
|
* interleaving bug-1415 describes on every run, not just occasionally.
|
|
*
|
|
* Against the pre-fix persist_register_pending_apply(), this reproduces the
|
|
* permanent hang byte for byte: the registration links onto pending_applies
|
|
* after the walk has already passed this lsn for good (the walk visits each
|
|
* lsn exactly once), so nothing ever consumes it — the hand-off (this
|
|
* test's stand-in for jobsubmit_apply_complete / the client reply) never
|
|
* fires, and the stand-in resource (this test's stand-in for the leaked
|
|
* jobsubmit_async_ctx_t / endpoint ref) is never released. Against the
|
|
* self-healing fix, persist_register_pending_apply() detects the
|
|
* already-applied lsn and fires the hand-off immediately and synchronously
|
|
* instead of linking it, so the test passes: the hand-off fires exactly
|
|
* once, the resource is released exactly once, and no registration is left
|
|
* dangling on pending_applies. */
|
|
|
|
#define TPR_BARRIER_TIMEOUT_MS 3000
|
|
#define TPR_WAIT_TIMEOUT_MS 8000
|
|
|
|
typedef struct {
|
|
persist_t* p;
|
|
pthread_mutex_t lock;
|
|
pthread_cond_t cond;
|
|
int saw_failed_outcome;
|
|
int pump_won_race; /* 1 once confirmed the walk
|
|
* applied lsn before registration
|
|
* ran */
|
|
int apply_complete_fires; /* must end up == 1 */
|
|
uint64_t apply_complete_lsn;
|
|
void* fake_resource; /* stands in for
|
|
* jobsubmit_async_ctx_t / the
|
|
* endpoint ref -- released exactly
|
|
* once by the hand-off firing */
|
|
int resource_released; /* must end up == 1 */
|
|
} tpr_race_ctx_t;
|
|
|
|
static cluster_t g_tpr_c;
|
|
static commit_engine_t g_tpr_hce;
|
|
static vtaskmgr_t g_tpr_pool;
|
|
static int g_tpr_hep_a, g_tpr_hep_b;
|
|
static char g_tpr_tmpdir[256];
|
|
static job_submit_data_t g_tpr_seed_submit;
|
|
|
|
static job_t* tpr_seed_job(uint64_t jobid, uint32_t state)
|
|
{
|
|
xce_job_t jdef;
|
|
|
|
memset(&g_tpr_seed_submit, 0, sizeof(g_tpr_seed_submit));
|
|
g_tpr_seed_submit.queue = (char*)"q";
|
|
g_tpr_seed_submit.user = (char*)"tester";
|
|
|
|
memset(&jdef, 0, sizeof(jdef));
|
|
jdef.jobid = jobid;
|
|
jdef.state = state;
|
|
jdef.submit = &g_tpr_seed_submit;
|
|
|
|
return makeJob(g_tpr_c.jobmgr, &jdef);
|
|
}
|
|
|
|
static void tpr_setup(int n_vcpus)
|
|
{
|
|
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*)"tpr-probe") == NULL)
|
|
xstring_table_init(64);
|
|
|
|
if (!g_tap_jobobj_ready) {
|
|
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
|
|
g_tap_jobobj_ready = 1;
|
|
}
|
|
|
|
memset(&g_tpr_c, 0, sizeof(g_tpr_c));
|
|
g_tpr_c.myhost = &g_tpr_c.myhost_info;
|
|
g_tpr_c.myhost_info.name = (char*)"ctrl-test-race";
|
|
g_tpr_c.myhost_info.key = (char*)"";
|
|
|
|
TEST_ASSERT_EQUAL_INT(0,
|
|
xce_hook_init(&g_tpr_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
|
|
|
|
g_tpr_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
|
|
g_tpr_c.qmgr = makeInstance(&xkvmgr_type, &str_params);
|
|
g_tpr_c.workermgr = makeInstance(&xkvmgr_type, &str_params);
|
|
g_tpr_c.resmgr = makeInstance(&xkvmgr_type, &str_params);
|
|
TEST_ASSERT_NOT_NULL(g_tpr_c.jobmgr);
|
|
TEST_ASSERT_NOT_NULL(g_tpr_c.qmgr);
|
|
TEST_ASSERT_NOT_NULL(g_tpr_c.workermgr);
|
|
TEST_ASSERT_NOT_NULL(g_tpr_c.resmgr);
|
|
zarray_init(&g_tpr_c.sharedinst, 16, NULL);
|
|
|
|
persist_config_t cfg = persist_config_defaults();
|
|
cfg.wal_sync_ms = 5000;
|
|
cfg.sync_timeout_ms = 2000;
|
|
|
|
snprintf(g_tpr_tmpdir, sizeof(g_tpr_tmpdir), "/tmp/xce_tpr62_XXXXXX");
|
|
TEST_ASSERT_NOT_NULL(mkdtemp(g_tpr_tmpdir));
|
|
persist_set_workdir(&g_tpr_c.persist, g_tpr_tmpdir, "ctrl");
|
|
TEST_ASSERT_EQUAL_INT(0, persist_init(&g_tpr_c.persist, &cfg, &g_tpr_c));
|
|
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_tpr_c.persist));
|
|
|
|
/* 3-member cluster: 2 stub standbys, engine wired, ACKs withheld until
|
|
* this test explicitly delivers one via repl_on_ack — mirrors
|
|
* tap_setup exactly, up to this point. */
|
|
TEST_ASSERT_TRUE(
|
|
repl_add_standby(&g_tpr_c.persist.repl, &g_tpr_hep_a, "sb_a", 7600) >= 0);
|
|
TEST_ASSERT_TRUE(
|
|
repl_add_standby(&g_tpr_c.persist.repl, &g_tpr_hep_b, "sb_b", 7601) >= 0);
|
|
TEST_ASSERT_EQUAL_INT(0,
|
|
commit_engine_init(&g_tpr_hce, &g_tpr_c.persist, 3, n_vcpus));
|
|
repl_set_commit_engine(&g_tpr_c.persist.repl, &g_tpr_hce);
|
|
repl_broadcast_send_hook = tap_drop_send_hook;
|
|
|
|
/* The one deliberate divergence from every other fixture in this file:
|
|
* a REAL multi-vcpu pool, started (vtaskmgr_start -- unlike
|
|
* test_commit_admission_pool.c's admission-only fixture, this test
|
|
* needs actual vtask execution on real worker threads), wired via
|
|
* commit_set_pool so commit_resolve_pending()'s two dispatches run as
|
|
* genuine concurrent vtasks instead of the inline fallback. */
|
|
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&g_tpr_pool, n_vcpus));
|
|
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&g_tpr_pool));
|
|
commit_set_pool(&g_tpr_hce, &g_tpr_pool);
|
|
}
|
|
|
|
static void tpr_teardown(void)
|
|
{
|
|
repl_broadcast_send_hook = NULL;
|
|
commit_engine_fini(&g_tpr_hce);
|
|
vtaskmgr_fini(&g_tpr_pool);
|
|
persist_stop(&g_tpr_c.persist);
|
|
persist_fini(&g_tpr_c.persist);
|
|
|
|
zarray_fini(&g_tpr_c.sharedinst);
|
|
if (g_tpr_c.resmgr) dereference(g_tpr_c.resmgr);
|
|
if (g_tpr_c.workermgr) dereference(g_tpr_c.workermgr);
|
|
if (g_tpr_c.qmgr) dereference(g_tpr_c.qmgr);
|
|
if (g_tpr_c.jobmgr) dereference(g_tpr_c.jobmgr);
|
|
xce_hook_fini(&g_tpr_c.hooksys);
|
|
}
|
|
|
|
static void tpr_wait_durable(uint64_t lsn, uint32_t deadline_ms)
|
|
{
|
|
for (uint32_t waited = 0; waited < deadline_ms; waited += 5) {
|
|
if (atomic_load(&g_tpr_c.persist.durable_lsn) >= lsn)
|
|
return;
|
|
usleep(5000);
|
|
}
|
|
}
|
|
|
|
/* Task B's hand-off target: this test's stand-in for jobsubmit_apply_complete.
|
|
* Fires exactly once when persist_register_pending_apply()'s self-healing
|
|
* check invokes it inline (the fix), or would fire never (pre-fix hang) if
|
|
* the walk had already orphaned the registration. */
|
|
static void tpr_apply_complete(void* ctx, uint64_t lsn)
|
|
{
|
|
tpr_race_ctx_t* rctx = (tpr_race_ctx_t*)ctx;
|
|
|
|
pthread_mutex_lock(&rctx->lock);
|
|
rctx->apply_complete_fires++;
|
|
rctx->apply_complete_lsn = lsn;
|
|
if (rctx->fake_resource != NULL) {
|
|
free(rctx->fake_resource);
|
|
rctx->fake_resource = NULL;
|
|
rctx->resource_released++;
|
|
}
|
|
pthread_cond_broadcast(&rctx->cond);
|
|
pthread_mutex_unlock(&rctx->lock);
|
|
}
|
|
|
|
/* Task A stand-in for jobsubmit_commit_continuation: deterministically
|
|
* forces the exact bad interleaving bug-1415 describes (see this test's
|
|
* header comment) by blocking, on this vtask's OWN worker thread, until the
|
|
* apply pump (Task B, running concurrently on a separate vcpu) has already
|
|
* applied `lsn` -- THEN registering, reproducing a genuinely late
|
|
* persist_register_pending_apply() call. */
|
|
static void tpr_commit_continuation(void* ctx, uint64_t lsn,
|
|
commit_outcome_t outcome, int rc)
|
|
{
|
|
tpr_race_ctx_t* rctx = (tpr_race_ctx_t*)ctx;
|
|
|
|
(void)rc;
|
|
|
|
if (outcome != COMMIT_OUTCOME_COMMITTED) {
|
|
pthread_mutex_lock(&rctx->lock);
|
|
rctx->saw_failed_outcome = 1;
|
|
pthread_cond_broadcast(&rctx->cond);
|
|
pthread_mutex_unlock(&rctx->lock);
|
|
return;
|
|
}
|
|
|
|
uint32_t waited_ms = 0;
|
|
while (persist_last_applied_lsn(rctx->p) < lsn &&
|
|
waited_ms < TPR_BARRIER_TIMEOUT_MS) {
|
|
usleep(2000);
|
|
waited_ms += 2;
|
|
}
|
|
|
|
pthread_mutex_lock(&rctx->lock);
|
|
rctx->pump_won_race = (persist_last_applied_lsn(rctx->p) >= lsn);
|
|
pthread_mutex_unlock(&rctx->lock);
|
|
|
|
/* The exact call bug-1415 describes as arriving too late -- the apply
|
|
* pump has already visited (and will never revisit) this lsn. */
|
|
persist_register_pending_apply(rctx->p, lsn, tpr_apply_complete, ctx);
|
|
}
|
|
|
|
static void test_apply_pump_late_pending_apply_registration_self_heals(void)
|
|
{
|
|
const int N_VCPUS = 4; /* > 1 is the whole point -- see header comment */
|
|
|
|
tpr_setup(N_VCPUS);
|
|
|
|
job_t* j = tpr_seed_job(950, JOB_STAT_PEND);
|
|
int seeded_ok = (j != NULL);
|
|
|
|
tpr_race_ctx_t rctx;
|
|
memset(&rctx, 0, sizeof(rctx));
|
|
pthread_mutex_init(&rctx.lock, NULL);
|
|
pthread_cond_init(&rctx.cond, NULL);
|
|
rctx.p = &g_tpr_c.persist;
|
|
rctx.fake_resource = malloc(64);
|
|
int resource_alloc_ok = (rctx.fake_resource != NULL);
|
|
|
|
int rc1 = persist_submit_job_fact(&g_tpr_c.persist, WAL_JOB_START, 950,
|
|
JOB_STAT_RUN);
|
|
uint64_t lsn = g_tpr_c.persist.current_lsn - 1;
|
|
|
|
/* Register the commit-outcome continuation directly against the commit
|
|
* engine -- the same mechanism jobsubmit_commit_continuation uses
|
|
* (commit_submit_async), standing in for it at this layer since this
|
|
* test's boundary is commit.c/persist.c, not job.c. */
|
|
int async_rc = commit_submit_async(&g_tpr_hce, lsn, tpr_commit_continuation,
|
|
&rctx);
|
|
|
|
tpr_wait_durable(lsn, 2000);
|
|
/* Majority ACK (self + 1 of 2 standbys = 2 of 3 peers) drives
|
|
* commit_on_ack -> commit_advance_locked -> commit_resolve_pending,
|
|
* which dispatches BOTH the continuation (Task A) and the apply-pump
|
|
* kick (Task B) as real vtasks onto the pool wired above. */
|
|
repl_on_ack(&g_tpr_c.persist.repl, &g_tpr_hep_a, lsn);
|
|
|
|
/* Bounded poll for the hand-off -- the key assertion: the client-facing
|
|
* hand-off must still fire even though the walk (Task B) won the race
|
|
* and applied lsn before the continuation (Task A) ever registered. */
|
|
struct timespec deadline;
|
|
clock_gettime(CLOCK_REALTIME, &deadline);
|
|
deadline.tv_sec += (TPR_WAIT_TIMEOUT_MS / 1000);
|
|
deadline.tv_nsec += (long)(TPR_WAIT_TIMEOUT_MS % 1000) * 1000000L;
|
|
if (deadline.tv_nsec >= 1000000000L) {
|
|
deadline.tv_sec += 1;
|
|
deadline.tv_nsec -= 1000000000L;
|
|
}
|
|
|
|
pthread_mutex_lock(&rctx.lock);
|
|
while (rctx.apply_complete_fires == 0) {
|
|
if (pthread_cond_timedwait(&rctx.cond, &rctx.lock, &deadline) != 0)
|
|
break;
|
|
}
|
|
int fires = rctx.apply_complete_fires;
|
|
int released = rctx.resource_released;
|
|
int pump_won_race = rctx.pump_won_race;
|
|
int saw_failed = rctx.saw_failed_outcome;
|
|
uint64_t fired_lsn = rctx.apply_complete_lsn;
|
|
pthread_mutex_unlock(&rctx.lock);
|
|
|
|
int has_pending_after = persist_has_pending_apply(&g_tpr_c.persist, lsn);
|
|
|
|
if (rctx.fake_resource != NULL) {
|
|
/* Only reachable pre-fix (orphaned forever, never released by
|
|
* tpr_apply_complete) -- free here so THIS test never leaks
|
|
* regardless of which way the assertions below come out. */
|
|
free(rctx.fake_resource);
|
|
rctx.fake_resource = NULL;
|
|
}
|
|
|
|
pthread_mutex_destroy(&rctx.lock);
|
|
pthread_cond_destroy(&rctx.cond);
|
|
|
|
tpr_teardown();
|
|
|
|
TEST_ASSERT_TRUE(seeded_ok);
|
|
TEST_ASSERT_TRUE(resource_alloc_ok);
|
|
TEST_ASSERT_EQUAL_INT(0, rc1);
|
|
TEST_ASSERT_EQUAL_INT(0, async_rc);
|
|
TEST_ASSERT_FALSE_MESSAGE(saw_failed,
|
|
"the entry must genuinely commit -- a FAILED outcome would mean "
|
|
"this test never exercises the race it targets");
|
|
TEST_ASSERT_TRUE_MESSAGE(pump_won_race,
|
|
"the forced barrier must confirm the apply pump actually applied "
|
|
"this lsn BEFORE the continuation registered -- otherwise this test "
|
|
"is not proving what it claims");
|
|
|
|
TEST_ASSERT_EQUAL_INT_MESSAGE(1, fires,
|
|
"bug-1415: a persist_register_pending_apply() call that arrives "
|
|
"after the apply pump has already applied its lsn must still fire "
|
|
"its hand-off exactly once (self-healing) -- 0 means a permanent "
|
|
"hang (the orphaned-registration bug), >1 means a double-fire");
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn, fired_lsn,
|
|
"the hand-off must fire for the correct lsn");
|
|
TEST_ASSERT_EQUAL_INT_MESSAGE(1, released,
|
|
"the hand-off's resource (standing in for jobsubmit_async_ctx_t / "
|
|
"the endpoint ref) must be released exactly once -- no leak, no "
|
|
"double-free");
|
|
TEST_ASSERT_EQUAL_INT_MESSAGE(0, has_pending_after,
|
|
"no orphaned registration may remain linked on pending_applies "
|
|
"after the hand-off fires");
|
|
|
|
printf("test_apply_pump_late_pending_apply_registration_self_heals: "
|
|
"PASS\n");
|
|
}
|
|
|
|
/* ── 6. Standby-ingest commit-then-apply (2026-07-10 fix) ────────────────
|
|
*
|
|
* Regression coverage for ha.c's standby WAL-replication ingest path.
|
|
* Before this fix, ha_apply_single_entry / ha_apply_term_entry_via_library
|
|
* called apply_wal_entry() directly the instant a replicated entry was
|
|
* durably persisted — materializing the job into the job table regardless
|
|
* of whether the sending primary ever reached cluster-wide majority
|
|
* commit. If that primary died before reaching majority, a later
|
|
* (term,lsn) conflict truncate would remove the WAL bytes with no
|
|
* rollback of the already-materialized entity, leaving a permanent
|
|
* phantom job.
|
|
*
|
|
* The fix decouples durability from materialization: ha_apply_single_entry
|
|
* / ha_apply_term_entry_via_library now only persist + ACK. Entity
|
|
* materialization happens later, only once THIS standby's own
|
|
* persist_t.committed_lsn actually covers the entry — driven by
|
|
* ha_standby_advance_committed_and_apply (clamped to
|
|
* min(learned_committed_lsn, local durable tip)) via
|
|
* persist_apply_pump_kick -> persist_apply_pump_run -> apply_committed_entry
|
|
* (the same seam the primary's own online-apply/replay use).
|
|
*
|
|
* These tests drive the REAL standby ingest seam — ha_apply_wal_catchup_entry,
|
|
* the exact function every incoming WAL replication message routes through —
|
|
* directly against this file's real jobmgr/persist fixture, so "not yet
|
|
* visible" / "now visible" is observed via tap_find_job, the same job-table
|
|
* lookup every other test in this file uses, not a synthetic stand-in.
|
|
* g_tap_c.ha.raft_c is never set to &g_tap_c in this fixture, so
|
|
* ha_apply_wal_catchup_entry always takes the documented term-blind
|
|
* direct-persist bypass (ha_apply_single_entry) regardless of the entries'
|
|
* term — the same reachable-and-safe-by-guard path
|
|
* test_6_2_real_senders_never_produce_termblind_shape (test_ha_core_safety.c)
|
|
* documents. */
|
|
|
|
/* Build one WAL_JOB_SUBMIT sub-entry for jobid at lsn, via the same capnp
|
|
* encoding tap_encode_job already uses for the live command-entry test
|
|
* above. Caller owns *payload_out (free() after the entry is no longer
|
|
* needed) and must keep it alive across the ha_apply_wal_catchup_entry
|
|
* call — persist_wal_write_replicated copies the bytes to disk, so the
|
|
* pointer's lifetime need not outlive the call, but freeing before the
|
|
* call would read stale data mid-encode. */
|
|
static wal_catchup_entry_t tas_build_submit_entry(uint64_t lsn, uint64_t jobid,
|
|
uint32_t state,
|
|
uint8_t** payload_out)
|
|
{
|
|
memset(&g_tap_seed_submit, 0, sizeof(g_tap_seed_submit));
|
|
g_tap_seed_submit.queue = (char*)"q";
|
|
g_tap_seed_submit.user = (char*)"tester";
|
|
|
|
xce_job_t jdef;
|
|
memset(&jdef, 0, sizeof(jdef));
|
|
jdef.jobid = jobid;
|
|
jdef.state = state;
|
|
jdef.submit = &g_tap_seed_submit;
|
|
|
|
uint8_t* payload = NULL;
|
|
size_t plen = 0;
|
|
TEST_ASSERT_EQUAL_INT(0, tap_encode_job(&jdef, &payload, &plen));
|
|
|
|
wal_catchup_entry_t sub;
|
|
memset(&sub, 0, sizeof(sub));
|
|
sub.lsn = lsn;
|
|
sub.entry_type = WAL_JOB_SUBMIT;
|
|
sub.raw_payload = payload;
|
|
sub.raw_payload_len = (uint32_t)plen;
|
|
/* term == 0 deliberately: see the file-header comment above — this
|
|
* fixture always exercises the term-blind direct-persist bypass. */
|
|
|
|
*payload_out = payload;
|
|
return sub;
|
|
}
|
|
|
|
/* (a) Positive case: persisted-but-uncommitted stays invisible; a
|
|
* heartbeat-learned commit position materializes it. */
|
|
static void test_standby_apply_deferred_until_committed_then_materializes(
|
|
void)
|
|
{
|
|
tap_setup();
|
|
|
|
const uint64_t lsn = 1; /* first entry in a fresh persist_t */
|
|
const uint64_t jobid = 960;
|
|
|
|
uint8_t* payload = NULL;
|
|
wal_catchup_entry_t sub = tas_build_submit_entry(lsn, jobid, JOB_STAT_PEND,
|
|
&payload);
|
|
wal_catchup_entry_t* sub_arr[1] = { &sub };
|
|
wal_catchup_entry_t top;
|
|
memset(&top, 0, sizeof(top));
|
|
top.n_entries = 1;
|
|
top.entries = sub_arr;
|
|
|
|
/* Ingest via the REAL standby seam: persist + ACK only, no apply. */
|
|
int rc = ha_apply_wal_catchup_entry(&g_tap_c.ha, &g_tap_c.persist, &top,
|
|
NULL, &g_tap_c);
|
|
|
|
uint64_t current_lsn_after_persist = g_tap_c.persist.current_lsn;
|
|
uint64_t applied_lsn_after_persist =
|
|
atomic_load(&g_tap_c.ha.applied_lsn);
|
|
uint64_t committed_after_persist = atomic_load(&g_tap_c.persist.committed_lsn);
|
|
job_t* j_before_commit = tap_find_job(jobid);
|
|
|
|
/* Learn the commit position the way a real heartbeat does -- this is
|
|
* what drives the commit-then-apply wiring (ha_consume_heartbeat_committed
|
|
* -> ha_standby_advance_committed_and_apply -> persist_apply_pump_kick). */
|
|
atomic_store(&g_tap_c.ha.role, (int32_t)CTRL_STANDBY);
|
|
|
|
heartbeat_msg_t hb;
|
|
memset(&hb, 0, sizeof(hb));
|
|
hb.term = 0;
|
|
hb.committed_lsn = lsn;
|
|
ha_consume_heartbeat_committed(&g_tap_c, &hb);
|
|
|
|
/* See tap_wait_applied's doc comment: the walk may complete on a
|
|
* different thread than this call. */
|
|
tap_wait_applied(lsn, 2000);
|
|
|
|
uint64_t committed_after_hb = atomic_load(&g_tap_c.persist.committed_lsn);
|
|
job_t* j_after_commit = tap_find_job(jobid);
|
|
uint32_t j_state_after = j_after_commit ? j_after_commit->job.state : 0;
|
|
uint64_t last_applied_final = persist_last_applied_lsn(&g_tap_c.persist);
|
|
|
|
free(payload);
|
|
tap_teardown();
|
|
|
|
TEST_ASSERT_EQUAL_INT(0, rc);
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn + 1, current_lsn_after_persist,
|
|
"the entry must be durably persisted by the standby ingest seam");
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn, applied_lsn_after_persist,
|
|
"applied_lsn (the held/persisted watermark) advances on receipt -- "
|
|
"distinct from entity materialization");
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, committed_after_persist,
|
|
"committed_lsn must NOT advance just because the entry was "
|
|
"persisted -- nothing has told this standby the entry reached "
|
|
"cluster majority yet");
|
|
TEST_ASSERT_NULL_MESSAGE(j_before_commit,
|
|
"REGRESSION GUARD: a replicated entry must not materialize into "
|
|
"the job table before this standby's own committed_lsn covers it "
|
|
"-- materializing here (the old eager-apply behavior) is exactly "
|
|
"the phantom-job hazard this fix closes");
|
|
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn, committed_after_hb,
|
|
"the heartbeat-learned commit position must advance committed_lsn");
|
|
TEST_ASSERT_NOT_NULL_MESSAGE(j_after_commit,
|
|
"once committed_lsn covers the entry, the apply pump must "
|
|
"materialize it into the job table");
|
|
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_PEND, j_state_after);
|
|
TEST_ASSERT_EQUAL_UINT64(lsn, last_applied_final);
|
|
|
|
printf("test_standby_apply_deferred_until_committed_then_materializes: "
|
|
"PASS\n");
|
|
}
|
|
|
|
/* (b) Regression guard: an entry truncated away before ever being learned
|
|
* as committed must never leave a phantom job — even when a stale/racing
|
|
* heartbeat still advertises the old (now-truncated) committed position. */
|
|
static void test_standby_truncate_before_commit_leaves_no_phantom_job(void)
|
|
{
|
|
tap_setup();
|
|
|
|
const uint64_t lsn = 1;
|
|
const uint64_t jobid = 961;
|
|
const uint64_t cutoff_lsn = lsn - 1; /* truncate away lsn and above */
|
|
|
|
uint8_t* payload = NULL;
|
|
wal_catchup_entry_t sub = tas_build_submit_entry(lsn, jobid, JOB_STAT_PEND,
|
|
&payload);
|
|
wal_catchup_entry_t* sub_arr[1] = { &sub };
|
|
wal_catchup_entry_t top;
|
|
memset(&top, 0, sizeof(top));
|
|
top.n_entries = 1;
|
|
top.entries = sub_arr;
|
|
|
|
int rc = ha_apply_wal_catchup_entry(&g_tap_c.ha, &g_tap_c.persist, &top,
|
|
NULL, &g_tap_c);
|
|
|
|
job_t* j_before_truncate = tap_find_job(jobid);
|
|
|
|
/* The sending primary died before reaching majority: a new leader's
|
|
* conflict-truncate removes the entry before this standby ever learns
|
|
* it was committed. */
|
|
int trc = persist_truncate_wal_above(&g_tap_c.persist, cutoff_lsn);
|
|
uint64_t current_lsn_after_truncate = g_tap_c.persist.current_lsn;
|
|
|
|
/* A stale/racing heartbeat still carries the old (now-truncated)
|
|
* committed position -- ha_standby_advance_committed_and_apply's clamp
|
|
* to the local durable tip (current_lsn-1) must refuse to walk past
|
|
* what is actually on disk. */
|
|
atomic_store(&g_tap_c.ha.role, (int32_t)CTRL_STANDBY);
|
|
|
|
heartbeat_msg_t hb;
|
|
memset(&hb, 0, sizeof(hb));
|
|
hb.term = 0;
|
|
hb.committed_lsn = lsn;
|
|
ha_consume_heartbeat_committed(&g_tap_c, &hb);
|
|
|
|
/* There is nothing valid left to apply, so no walk will ever reach
|
|
* `lsn` -- a bounded settle (not a wait-for) is the correct shape here. */
|
|
usleep(50000);
|
|
|
|
uint64_t committed_after = atomic_load(&g_tap_c.persist.committed_lsn);
|
|
uint64_t last_applied_final = persist_last_applied_lsn(&g_tap_c.persist);
|
|
job_t* j_after = tap_find_job(jobid);
|
|
|
|
free(payload);
|
|
tap_teardown();
|
|
|
|
TEST_ASSERT_EQUAL_INT(0, rc);
|
|
TEST_ASSERT_NULL_MESSAGE(j_before_truncate,
|
|
"must not be materialized before commit (same guard as the "
|
|
"positive-case test)");
|
|
TEST_ASSERT_EQUAL_INT(0, trc);
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(cutoff_lsn + 1, current_lsn_after_truncate,
|
|
"the entry must actually be gone from the WAL after truncation");
|
|
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, committed_after,
|
|
"REGRESSION GUARD: the local-durable-tip clamp in "
|
|
"ha_standby_advance_committed_and_apply must refuse to advance "
|
|
"committed_lsn past a truncated-away entry, even when a stale "
|
|
"heartbeat still advertises it as committed");
|
|
TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, last_applied_final,
|
|
"nothing may be applied past the truncated tip");
|
|
TEST_ASSERT_NULL_MESSAGE(j_after,
|
|
"REGRESSION GUARD: no phantom job may appear -- the entry was "
|
|
"truncated before ever being committed, so it must never be "
|
|
"materialized");
|
|
|
|
printf("test_standby_truncate_before_commit_leaves_no_phantom_job: "
|
|
"PASS\n");
|
|
}
|
|
|
|
/* ── Test group entry point ───────────────────────────────────────────── */
|
|
|
|
void test_apply_pump(void)
|
|
{
|
|
UNITY_BEGIN();
|
|
RUN_TEST(test_apply_pump_kick_applies_committed_burst_in_ascending_lsn_order);
|
|
RUN_TEST(test_apply_pump_run_applies_mixed_submit_and_fact_entries);
|
|
RUN_TEST(test_apply_pump_run_halts_at_failed_lsn_without_advancing);
|
|
RUN_TEST(test_apply_pump_run_engages_live_fence_on_failed_lsn);
|
|
RUN_TEST(test_apply_pump_run_self_clears_fence_on_successful_retry);
|
|
RUN_TEST(test_apply_pump_guard_does_not_break_job_ctrl_live_append);
|
|
RUN_TEST(test_apply_pump_fail_racing_finish_merges_against_current_state);
|
|
RUN_TEST(test_apply_pump_late_pending_apply_registration_self_heals);
|
|
RUN_TEST(test_standby_apply_deferred_until_committed_then_materializes);
|
|
RUN_TEST(test_standby_truncate_before_commit_leaves_no_phantom_job);
|
|
UNITY_END();
|
|
}
|