wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

test(raft-apply-decoupling): mixed facts+commands quorum-loss catch-up ordering (task 6.4)

Browse source 
Extend test_apply_pump_integration.c with 2 tests, no production
changes. Test 3 interleaves commands and facts through a real
quorum-loss outage and proves strict original-log-order catch-up via
a sequence spy. Test 4 wires a real 4-vcpu vtaskmgr_t to drive 4
commands + 2 facts through one covering ACK, creating genuine N-way
concurrent races across task 6.2's self-healing fix rather than one
isolated forced pair -- stress-testing bug-1415's fix under a
realistic catch-up burst.

Closes out task-group 6 (ordered apply pump integration): the pump
is now live, hands off command replies correctly, and is proven
correct for isolated and mixed/bursty command+fact scenarios under
both normal and quorum-loss conditions.
This commit is contained in:
Rongsong Shen 2026-07-05 01:46:38 +00:00
parent 5f3b97a97b
commit 2de410c877
2 changed files with 613 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

5
.kiro/specs/raft-apply-decoupling/tasks.md
View file

File diff suppressed because one or more lines are too long

620
src/controller/testsuite/test_apply_pump_integration.c
View file

@ -1,4 +1,4 @@
/* test_apply_pump_integration.c — raft-apply-decoupling task 6.3
/* test_apply_pump_integration.c — raft-apply-decoupling tasks 6.3 and 6.4
*
* Integration coverage for command AND fact apply/notify timing through the
* SAME full pump (task 6.1's ordered walk + task 6.2's continuation
@ -6,6 +6,19 @@
* (which predated the pump and could only pin "no phantom job", not
* "notified only after commit").
*
* Tests 1-2 are task 6.3's (command+fact interleave through one commit; fact
* reactions withheld/restored across a genuine outage). Tests 3-4 are task
* 6.4's: a genuine quorum-loss-then-restore catch-up over a MIX of commands
* AND facts (Test 2 above is fact-only, by design — see task 6.1's
* Implementation Notes calling that out as non-duplicative of 6.4's
* charter), plus a real-multi-vcpu-pool burst variant that stress-tests task
* 6.2's bug-1415 self-heal (persist_register_pending_apply) under several
* concurrent command continuations racing the walk at once, not just the
* one forced interleaving test_apply_pump.c's own regression test covers.
* See Test 3 and Test 4's own header comments below for the full scope
* split and why NULL-pool determinism vs a real pool each earn their own
* test rather than one doing both jobs.
*
* Two things this file does NOT re-prove, because they are already covered
* elsewhere in this exact spec's own suite:
* - Pure command reply-after-visibility ordering for job submit —
@ -72,13 +85,16 @@
*
* Fixture mirrors test_apply_pump.c's tap_setup (real persist_init +
* persist_start over a private tmpdir, a real 3-member commit_engine_t with
* two stub standbys, ACKs delivered explicitly via repl_on_ack, commit_
* set_pool never called so the apply pump's NULL-pool inline fallback
* applies) plus test_commit_barrier_reply.c's reply-observation harness
* (a zeroed, no-iomgr requester endpoint; xce_reply's packets stay queued
* on ep->outq and are parsed with the raw capnp reader).
* two stub standbys, ACKs delivered explicitly via repl_on_ack) plus
* test_commit_barrier_reply.c's reply-observation harness (a zeroed,
* no-iomgr requester endpoint; xce_reply's packets stay queued on ep->outq
* and are parsed with the raw capnp reader). tpi_setup() takes a
* pool_n_vcpus parameter (task 6.4): 0 (Tests 1-3) leaves ce->pool == NULL
* so the apply pump's inline fallback applies, exactly as tasks 6.3's own
* tests always ran; > 0 (Test 4 only) wires a REAL vtaskmgr_t, mirroring
* test_apply_pump.c's task 6.2 tpr_setup.
*
* Both tests below follow this series' established disciplines: capture
* All four tests below follow this series' established disciplines: capture
* every assertion-relevant observable into locals, then teardown
* unconditionally, THEN assert (a failed TEST_ASSERT's longjmp must never
* skip teardown); poll for eventual apply-pump consistency rather than
@ -113,6 +129,7 @@
#include "persist.h"
#include "repl.h"
#include "request.h"
#include "vtaskmgr.h"
#include "wal_types.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
@ -128,6 +145,16 @@ static commit_engine_t g_tpi_hce;
static int g_tpi_hep_a, g_tpi_hep_b; /* dummy standby endpoints */
static char g_tpi_tmpdir[256];
/* task 6.4: a REAL multi-vcpu pool, wired only when a test explicitly asks
* for one via tpi_setup's pool_n_vcpus parameter -- every fixture through
* task 6.3 keeps ce->pool == NULL (see test_apply_pump.c's task 6.2 bug-1415
* regression test header for why that trivially serializes commit_resolve_
* pending()'s two independent dispatches and so cannot reproduce that race
* class). g_tpi_pool_active tracks whether tpi_teardown owes a
* vtaskmgr_fini this run. */
static vtaskmgr_t g_tpi_pool;
static int g_tpi_pool_active;
/* Test-injectable send hook (repl.c, XCE_TESTING) — keeps repl_broadcast
* from dereferencing the dummy standby endpoints above. */
extern int (*repl_broadcast_send_hook)(void *ep, wal_catchup_entry_t **entries,
@ -147,11 +174,23 @@ static int tpi_drop_send_hook(void *ep, wal_catchup_entry_t **entries,
static endpoint_t g_tpi_req_ep;
static ep_private_t g_tpi_req_priv;
/* task 6.4: bounded cap on how many individual replies tpi_reply_obs_t can
* distinguish by jobid/status -- separate from TPI_SPY_CAP below since the
* two arrays serve different structs; sized generously above the largest
* batch this file's tests submit. */
#define TPI_REPLY_CAP 8
typedef struct {
int n_replies;
int32_t last_status;
int32_t last_data_which;
uint64_t last_jobid;
/* task 6.4: per-reply jobid/status, in receipt order, so a mixed-batch
* test can confirm EVERY submitted command got its own exactly-one
* reply rather than only inspecting the last one observed. Unused by
* tasks 6.3's tests, which only ever check n_replies/last_*. */
uint64_t jobids[TPI_REPLY_CAP];
int32_t statuses[TPI_REPLY_CAP];
} tpi_reply_obs_t;
static void tpi_scan_replies(endpoint_t *ep, tpi_reply_obs_t *obs)
@ -170,6 +209,12 @@ static void tpi_scan_replies(endpoint_t *ep, tpi_reply_obs_t *obs)
if (rq.body_which == Request_body_replymsg) {
struct RequestReply rr;
read_RequestReply(&rr, rq.body.replymsg);
if (obs->n_replies < TPI_REPLY_CAP) {
obs->jobids[obs->n_replies] =
(rr.data_which == RequestReply_data_jobid) ? rr.data.jobid
: 0;
obs->statuses[obs->n_replies] = rr.status;
}
obs->n_replies += 1;
obs->last_status = rr.status;
obs->last_data_which = rr.data_which;
@ -182,6 +227,26 @@ static void tpi_scan_replies(endpoint_t *ep, tpi_reply_obs_t *obs)
}
}
/* task 6.4: true iff `jobid` appears in obs's captured replies exactly once,
* with a success (0) status -- catches both a hang (0 occurrences, since a
* missing reply never appears) and a bug-1415-class double-fire (>1
* occurrence) with one call, for the mixed-batch tests below. */
static bool tpi_reply_has_exactly_one_success(const tpi_reply_obs_t *obs,
uint64_t jobid)
{
int count = 0;
int cap = obs->n_replies < TPI_REPLY_CAP ? obs->n_replies : TPI_REPLY_CAP;
for (int i = 0; i < cap; i++) {
if (obs->jobids[i] == jobid) {
if (obs->statuses[i] != 0)
return false;
count++;
}
}
return count == 1;
}
/* ── Reaction spy: stands in for sched.c's sched_job_update_fun /
* apiserver.c's apiserver_job_update_fun (the real, only-reachable-for-
* facts subscribers — see file header). Captures, at the EXACT moment the
@ -263,7 +328,13 @@ static void tpi_spy_reset(void)
memset(g_tpi_spy, 0, sizeof(g_tpi_spy));
}
static void tpi_setup(void)
/* pool_n_vcpus == 0 (every task-6.3 test, and task 6.4's deterministic
* ordering test) keeps ce->pool == NULL exactly as before -- no behavior
* change for any existing caller. pool_n_vcpus > 0 (task 6.4's burst test
* only) wires a REAL vtaskmgr_t, mirroring test_apply_pump.c's task 6.2
* tpr_setup, so commit_resolve_pending()'s two independent dispatches run
* as genuine concurrent vtasks instead of the inline fallback. */
static void tpi_setup(int pool_n_vcpus)
{
kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 32,
@ -326,6 +397,14 @@ static void tpi_setup(void)
repl_set_commit_engine(&g_tpi_c.persist.repl, &g_tpi_hce);
repl_broadcast_send_hook = tpi_drop_send_hook;
g_tpi_pool_active = 0;
if (pool_n_vcpus > 0) {
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&g_tpi_pool, pool_n_vcpus));
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&g_tpi_pool));
commit_set_pool(&g_tpi_hce, &g_tpi_pool);
g_tpi_pool_active = 1;
}
TEST_ASSERT_EQUAL_INT(0,
xce_add_hook(&g_tpi_c.hooksys, XCE_CONTROLLER_HOOK_JOB, tpi_reaction_spy,
&g_tpi_c));
@ -340,6 +419,13 @@ static void tpi_teardown(void)
tpi_scan_replies(&g_tpi_req_ep, &drain); /* free any leftover packets */
repl_broadcast_send_hook = NULL;
commit_engine_fini(&g_tpi_hce);
if (g_tpi_pool_active) {
/* Mirrors test_apply_pump.c's tpr_teardown ordering exactly:
* vtaskmgr_fini AFTER commit_engine_fini (so nothing can dispatch a
* new vtask onto a pool mid-teardown) but BEFORE persist_stop. */
vtaskmgr_fini(&g_tpi_pool);
g_tpi_pool_active = 0;
}
persist_stop(&g_tpi_c.persist);
persist_fini(&g_tpi_c.persist);
@ -408,6 +494,26 @@ static void tpi_wait_applied(uint64_t lsn, uint32_t deadline_ms)
}
}
/* task 6.4: waits until at least `want_n` replies have been observed on the
* requester endpoint, accumulating across repeated tpi_scan_replies calls
* (each call only drains packets newly enqueued since the last one). Needed
* specifically for the real-pool burst test: tpi_wait_applied only proves
* the WALK finished, not that every command's continuation hand-off has
* fired -- a hand-off that self-heals late (persist_register_pending_apply,
* bug-1415) runs on its OWN vtask thread, independently of and potentially
* after the walk. */
static void tpi_wait_n_replies(tpi_reply_obs_t *obs, int want_n,
uint32_t deadline_ms)
{
for (uint32_t waited = 0; waited < deadline_ms; waited += 5) {
tpi_scan_replies(&g_tpi_req_ep, obs);
if (obs->n_replies >= want_n)
return;
usleep(5000);
}
tpi_scan_replies(&g_tpi_req_ep, obs); /* one last drain past the deadline */
}
static void tpi_sleep_ms(uint32_t ms)
{
struct timespec ts;
@ -468,7 +574,7 @@ static void test_pump_command_and_fact_interleave_through_same_commit(void)
const uint64_t jobid_x = 9500; /* submitted via handle_jobsubmit */
const uint64_t jobid_y = 9501; /* pre-existing, receives a START fact */
tpi_setup();
tpi_setup(0);
job_t *jy = tpi_seed_job(jobid_y, JOB_STAT_PEND);
int seeded_ok = (jy != NULL);
@ -630,7 +736,7 @@ static void test_pump_fact_reactions_withheld_during_outage_then_fire_after_rest
{
const uint64_t jobid_z = 9600;
tpi_setup();
tpi_setup(0);
job_t *jz = tpi_seed_job(jobid_z, JOB_STAT_PEND);
int seeded_ok = (jz != NULL);
@ -751,6 +857,498 @@ static void test_pump_fact_reactions_withheld_during_outage_then_fire_after_rest
"after_restore: PASS\n");
}
/* ═══════════════════════════════════════════════════════════════════════
* Test 3 — task 6.4: MIXED facts AND commands through a genuine quorum-loss-
* then-restore window, proving strict original-log-order catch-up.
* ═══════════════════════════════════════════════════════════════════════
*
* Distinguishing scope vs this file's Test 2 above (task 6.3) and
* test_commit_barrier_reply.c's test_fact_state_unchanged_throughout_
* quorum_loss_outage (task 4.2), both of which drive a genuine repl_remove_
* standby outage with FACTS ONLY: this test interleaves THREE job-submit
* COMMANDS with TWO facts, five entries total, appended in COMMAND, FACT,
* COMMAND, FACT, COMMAND order specifically so a bug that grouped all
* commands before all facts (or vice versa) -- rather than genuinely
* preserving append order -- would be caught by the strict per-slot
* sequence assertion below, not merely "everything eventually applied, in
* some order." Requirements 6.3 (retained entries commit/apply in original
* log order) and 7.3 (never applies ahead of an earlier-committing entry)
* are the direct target; 9.4 is checked negatively -- the outage must never
* spuriously resolve a registered command continuation as a FAILURE (the
* demotion/truncation path that requirement's other clause governs), only
* ever as a genuine, later, COMMITTED success once quorum returns.
*
* Uses the standard NULL-pool fixture (tpi_setup(0)) deliberately, exactly
* like this file's other two tests: with ce->pool == NULL, commit_resolve_
* pending()'s continuation dispatch and the apply-pump kick both fall back
* to their synchronous inline path (dispatch_continuation, apply_pump_drain
* -- see commit.c), so the ENTIRE catch-up (register, apply, notify, reply)
* happens deterministically on this test's own thread inside the single
* repl_on_ack() call below. That determinism is what makes the strict,
* order-sensitive sequence assertion below meaningful rather than flaky --
* Test 4 below separately re-verifies the same ordering invariant survives
* genuine concurrency, and additionally stress-tests task 6.2's bug-1415
* self-heal under a real multi-entry burst. */
static void test_pump_mixed_facts_and_commands_catchup_original_order(void)
{
const uint64_t jobid_a = 9700; /* command 1 */
const uint64_t jobid_w = 9701; /* pre-existing, receives both facts */
const uint64_t jobid_b = 9702; /* command 2 */
const uint64_t jobid_c = 9703; /* command 3 */
tpi_setup(0);
job_t *jw = tpi_seed_job(jobid_w, JOB_STAT_PEND);
int seeded_ok = (jw != NULL);
/* Genuine quorum loss -- same mechanism as this file's Test 2 (task
* 6.3) and test_commit_barrier_reply.c's task 4.2 test: both standbys
* removed, self alone is 1 of 3, never a majority. */
repl_remove_standby(&g_tpi_c.persist.repl, &g_tpi_hep_a);
repl_remove_standby(&g_tpi_c.persist.repl, &g_tpi_hep_b);
pthread_mutex_lock(&g_tpi_c.persist.lock);
uint64_t lsn_before = g_tpi_c.persist.current_lsn;
pthread_mutex_unlock(&g_tpi_c.persist.lock);
/* Append order: COMMAND, FACT, COMMAND, FACT, COMMAND -- deliberately
* NOT grouped by type. */
tpi_arrange_jobsubmit(jobid_a);
int submit_a_rc =
handle_jobsubmit(&g_tpi_c, &g_tpi_req_ep, &g_tpi_jobsub_req);
static unsigned int w_pids[1] = { 7201 };
job_pidinfo_t w_start = { .jobid = jobid_w,
.npids = 1,
.pids = &w_pids[0] };
int start_rc = on_job_start(&g_tpi_c, &w_start);
tpi_arrange_jobsubmit(jobid_b);
int submit_b_rc =
handle_jobsubmit(&g_tpi_c, &g_tpi_req_ep, &g_tpi_jobsub_req);
xce_jobfinish_t w_finish = { .jobid = jobid_w, .exit_code = 0 };
int finish_rc = on_job_finish(&g_tpi_c, &w_finish);
tpi_arrange_jobsubmit(jobid_c);
int submit_c_rc =
handle_jobsubmit(&g_tpi_c, &g_tpi_req_ep, &g_tpi_jobsub_req);
pthread_mutex_lock(&g_tpi_c.persist.lock);
uint64_t watermark = g_tpi_c.persist.current_lsn - 1;
pthread_mutex_unlock(&g_tpi_c.persist.lock);
/* ── Outage window: poll repeatedly across real time -- all five
* entries are durably appended and retained, but structurally cannot
* commit while both standbys remain disconnected (Req 6.1). ── */
enum { OUTAGE_POLLS = 5, OUTAGE_INTERVAL_MS = 20 };
bool outage_a_absent[OUTAGE_POLLS];
bool outage_b_absent[OUTAGE_POLLS];
bool outage_c_absent[OUTAGE_POLLS];
uint32_t outage_w_state[OUTAGE_POLLS];
int outage_spy_n[OUTAGE_POLLS];
uint64_t outage_committed[OUTAGE_POLLS];
int outage_replies[OUTAGE_POLLS];
for (int i = 0; i < OUTAGE_POLLS; i++) {
tpi_sleep_ms(OUTAGE_INTERVAL_MS);
outage_a_absent[i] = (tpi_find_job(jobid_a) == NULL);
outage_b_absent[i] = (tpi_find_job(jobid_b) == NULL);
outage_c_absent[i] = (tpi_find_job(jobid_c) == NULL);
outage_w_state[i] = seeded_ok ? jw->job.state : 0;
outage_spy_n[i] = g_tpi_spy_n;
outage_committed[i] = atomic_load(&g_tpi_c.persist.committed_lsn);