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

test(ha-convergence-recovery): RED ACK-driven catch-up continuation tests + seam (task 4.1)

Browse source
This commit is contained in:
Rongsong Shen 2026-06-14 05:16:54 +00:00
parent 63c12461a5
commit c2221077ff
6 changed files with 566 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -54,6 +54,7 @@ extern void test_commit_barrier_rollout(void);
extern void test_commit_barrier_reply(void);
extern void test_egress_filter(void);
extern void test_demotion_convergence(void);
extern void test_convergence_verification(void);
void setUp(void) { return; }
@ -177,6 +178,7 @@ int main(int argc, char* argv[])
TEST(test_commit_barrier_reply);
TEST(test_egress_filter);
TEST(test_demotion_convergence);
TEST(test_convergence_verification);
return UNITY_END();
}

483
src/controller/testsuite/test_convergence_verification.c Normal file
View file

@ -0,0 +1,483 @@
#define _GNU_SOURCE
/*
* test_convergence_verification.c — RED-phase tests for ACK-driven catch-up
* continuation, single-flight, and parity re-inclusion
* (ha-convergence-recovery task 4.1).
*
* Requirements (this spec's IDs — see requirements.md):
* 5.1 — when the primary receives a WAL ACK from a standby that remains BEHIND
* the replication target, it continues streaming the missing entries to
* that standby WITHOUT waiting for the next heartbeat exchange (the
* continuation streams [acked_lsn+1, target] and dispatches no
* heartbeat/stateReport replication action of its own).
* 5.2 — while a catch-up continuation for a standby is already in flight, the
* primary does NOT start a duplicate continuation for that standby
* (per-standby single-flight guard); once the in-flight continuation
* completes, the next lagging ACK schedules again.
* 5.3 — the heartbeat-driven replication evaluation remains in place as the
* fallback initiator of catch-up AND the initiator of snapshot-transfer
* decisions: ha_evaluate_repl_action still returns HA_REPL_STREAM_WAL /
* HA_REPL_SEND_SNAPSHOT for the same inputs (decision boundary pin).
* 7.1 — when a lagging standby catches up to the replication boundary (a WAL
* ACK at/above target), the primary re-includes it in live replication
* via repl_complete_catchup, which clears the standby's in_catch_up.
*
* RED status against the current tree: the production walAck branch
* (repl_ack_decode_and_dispatch, ha.c — the OPCode_reply /
* RequestReply_data_walackreplymsg case) only calls repl_on_ack; it derives no
* target, schedules no continuation, and never calls repl_complete_catchup
* (which has NO production caller, repl.c). The task-4.1 test seam
* ha_walack_continuation_for_test is a RED scaffold (ha_convergence.c) that
* records the ACK only. The continuation / single-flight / parity cases below
* therefore FAIL until task 4.2 lands the real D6 decision; the 5.3 decision-pin
* case exercises existing behavior and PASSES now (pinning the boundary so 4.2
* cannot disturb the heartbeat path).
*
* Harness: a partial-but-real cluster_t (real persist over a private tmpdir,
* a wired commit engine, stub standby registry, un-started delaymgr) mirroring
* test_demotion_convergence.c's dc_setup, plus the catch-up streaming count seam
* ha_repl_stream_wal_hook and a dispatch-observer seam to prove the continuation
* takes no heartbeat path.
*
* AAA structure throughout.
*/
#include <assert.h>
#include <errno.h>
#include <inttypes.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <unity/unity.h>
#include "commit.h" /* commit_engine_t, commit_engine_init */
#include "delay.h" /* delaymgr_t, delay_task_t */
#include "ha.h" /* ha_walack_continuation_for_test, ha_evaluate_repl_action,
* ha_repl_stream_wal_hook, HA_REPL_* */
#include "ordq.h" /* ordq_item_t (drive the scheduled task) */
#include "persist.h" /* persist_t */
#include "repl.h" /* repl_add_standby / repl_set_commit_engine / standby slots */
#include "xce/cluster_def.h" /* full cluster_t layout */
#ifdef XCE_TESTING
/* ── Broadcast drop-hook (precedent: test_demotion_convergence.c) — the stub int
* endpoints registered via repl_add_standby are not real endpoint_t, so any
* repl_broadcast triggered by priming appends must not dereference ep->peer. */
extern int (*repl_broadcast_send_hook)(void *ep,
wal_catchup_entry_t **entries,
int n_entries);
static int cv_drop_send_hook(void *ep, wal_catchup_entry_t **entries,
int n_entries)
{
(void)ep; (void)entries; (void)n_entries;
return 0;
}
/* ── Catch-up streaming count seam (Req 5.1) ─────────────────────────────────
* The ACK-driven continuation must stream the missing range through the existing
* catch-up streaming path. ha_repl_stream_wal_hook is the established XCE_TESTING
* seam for that path (config_repl.c ha_stream_wal_to_standby / ha.c
* HA_REPL_STREAM_WAL). We count invocations and capture the [start, target]
* range so the continuation's bounds can be asserted. */
static atomic_int g_stream_calls;
static atomic_uint_least64_t g_stream_last_standby_lsn;
static atomic_uint_least64_t g_stream_last_target_lsn;
static int cv_stream_hook(void *ep, const char *wal_path,
uint64_t standby_lsn, uint64_t committed_lsn)
{
(void)ep; (void)wal_path;
atomic_fetch_add(&g_stream_calls, 1);
atomic_store(&g_stream_last_standby_lsn, standby_lsn);
atomic_store(&g_stream_last_target_lsn, committed_lsn);
return 0;
}
/* ── Cluster fixture ─────────────────────────────────────────────────────── */
static cluster_t g_cv; /* large — must not live on the test stack */
static int g_cv_eps[MAX_STANDBYS];
#define CV_PERSIST_NAME "ctrl-test"
static ha_config_t cv_ha_cfg(void)
{
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
};
return cfg;
}
/* Build g_cv: real persist over a fresh tmpdir, n_standbys stub standbys, a
* wired commit engine, an initialised-but-UNSTARTED delaymgr (so any task in
* tasks.head was armed by the code under test, not by a background runner).
* Role left at INIT (the walAck continuation is a primary-side concern but the
* decision under test needs only the persist watermark + repl registry, not the
* full promotion sequence). Leaves committed_lsn at 0; callers set it. */
static void cv_setup(const char *suffix, int n_peers, int n_standbys,
uint64_t appended_lsn, int n_vcpus)
{
char name[32];
char tmpdir[256];
memset(&g_cv, 0, sizeof(g_cv));
atomic_store(&g_stream_calls, 0);
atomic_store(&g_stream_last_standby_lsn, 0);
atomic_store(&g_stream_last_target_lsn, 0);
ha_config_t cfg = cv_ha_cfg();
TEST_ASSERT_EQUAL_INT(0, ha_init(&g_cv.ha, &cfg, 0));
g_cv.ha.phase2_done = 1;
g_cv.ha.n_peers = n_peers;
persist_config_t pcfg = persist_config_defaults();
pcfg.wal_sync_ms = 5000; /* slow background fsync — no races */
pcfg.sync_timeout_ms = 50; /* no real standby ACKs here */
snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_cv_%s_XXXXXX", suffix);
TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));
persist_set_workdir(&g_cv.persist, tmpdir, CV_PERSIST_NAME);
TEST_ASSERT_EQUAL_INT(0, persist_init(&g_cv.persist, &pcfg, &g_cv));
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_cv.persist));
/* Prime the appended watermark: current_lsn = appended_lsn + 1, so
* last_written = appended_lsn (the replication target's upper bound). */
pthread_mutex_lock(&g_cv.persist.lock);
g_cv.persist.current_lsn = appended_lsn + 1;
pthread_mutex_unlock(&g_cv.persist.lock);
atomic_store(&g_cv.persist.committed_lsn, 0);
for (int i = 0; i < n_standbys; i++) {
snprintf(name, sizeof(name), "cvsb%d", i);
TEST_ASSERT_TRUE_MESSAGE(
repl_add_standby(&g_cv.persist.repl, &g_cv_eps[i], name,
7600 + (uint32_t)i) >= 0,
"repl_add_standby must return a slot index >= 0");
}
TEST_ASSERT_EQUAL_INT(0,
commit_engine_init(&g_cv.commit, &g_cv.persist, n_peers, n_vcpus));
repl_set_commit_engine(&g_cv.persist.repl, &g_cv.commit);
repl_broadcast_send_hook = cv_drop_send_hook;
ha_repl_stream_wal_hook = cv_stream_hook;
TEST_ASSERT_EQUAL_INT(0, delaymgr_init(&g_cv.delaymgr, 100, NULL));
}
static void cv_teardown(void)
{
repl_broadcast_send_hook = NULL;
ha_repl_stream_wal_hook = NULL;
commit_engine_fini(&g_cv.commit);
delaymgr_fini(&g_cv.delaymgr);
persist_stop(&g_cv.persist);
persist_fini(&g_cv.persist);
repl_fini(&g_cv.persist.repl);
ha_fini(&g_cv.ha);
/* tmpdir left on disk — /tmp is ephemeral */
}
/* Count pending delaymgr tasks by walking the ordq head chain (read-only;
* single-threaded test, delaymgr never started). */
static int cv_task_count(const delaymgr_t *mgr)
{
int count = 0;
const ordq_item_t *it = mgr->tasks.head;
while (it != NULL) {
count++;
it = it->next;
}
return count;
}
/* Run the single pending delaymgr one-shot task synchronously by invoking its
* stored entry(args), then return the entry's rc. Returns CV_NO_TASK if there
* is no pending task (RED: nothing was scheduled). DEQUEUES + frees, mirroring
* the real delaymgr worker (delay.c) and test_demotion_convergence.c's
* dc_run_pending_task — required so a second call after a body that frees its own
* ctx does not re-run a freed ctx (heap-UAF). */
#define CV_NO_TASK 0x7fffffff
static int cv_run_pending_task(delaymgr_t *mgr)
{
ordq_item_t *it = ordq_get(&mgr->tasks, NULL);
if (it == NULL || it->data == NULL) {
if (it != NULL)
free(it);
return CV_NO_TASK;
}
delay_task_t *t = (delay_task_t *)it->data;
if (t->entry == NULL) {
free(t);
free(it);
return CV_NO_TASK;
}
int rc = t->entry(t->args); /* task body frees its own args ctx */
free(t);
free(it);
return rc;
}
/* Read the per-standby in_catch_up flag (repl registry slot identified by ep). */
static int cv_standby_in_catch_up(repl_ep_t ep)
{
replicator_t *r = &g_cv.persist.repl;
int result = -1;
mylock_lock(&r->lock);
for (int i = 0; i < MAX_STANDBYS; i++) {
if (r->standbys[i].ep == ep) {
result = atomic_load_explicit(&r->standbys[i].in_catch_up,
memory_order_acquire);
break;
}
}
mylock_unlock(&r->lock);
return result;
}
/* ── Tests ──────────────────────────────────────────────────────────────── */
/* 5.1: a walAck with acked_lsn < target = max(committed_lsn, last_written)
* schedules EXACTLY ONE continuation that streams [acked_lsn+1, target] without
* any intervening heartbeat/stateReport dispatch.
* RED: the walAck continuation scaffold schedules nothing → no task, the stream
* hook is never called. */
static void test_walack_behind_target_schedules_one_continuation(void)
{
/* appended_lsn=9 → last_written=9; committed_lsn=4 → target=max(4,9)=9. */
cv_setup("behind", /*n_peers=*/3, /*n_standbys=*/1,
/*appended_lsn=*/9, /*n_vcpus=*/8);
atomic_store(&g_cv.persist.committed_lsn, 4);
const int peer_idx = 0;
const uint64_t acked_lsn = 3; /* behind target 9 */
const uint64_t target = 9;
int tasks_before = cv_task_count(&g_cv.delaymgr);
/* Act: standby acknowledges lsn=3 while target=9. */
int rc = ha_walack_continuation_for_test(&g_cv, &g_cv_eps[peer_idx],
peer_idx, acked_lsn);
(void)rc;
/* A continuation must have been scheduled OFF the ACK path (exactly one). */
int tasks_after = cv_task_count(&g_cv.delaymgr);
TEST_ASSERT_EQUAL_INT_MESSAGE(tasks_before + 1, tasks_after,
"a walAck behind target must schedule exactly one catch-up continuation "
"(Req 5.1)");
/* No streaming may have happened on the ACK path itself (it defers). */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_stream_calls),
"the walAck branch must NOT stream inline — streaming runs in the "
"deferred continuation (Req 5.1, iomgr-safe)");
/* Run the scheduled continuation; it must stream [acked_lsn+1, target]. */
int trc = cv_run_pending_task(&g_cv.delaymgr);
TEST_ASSERT_NOT_EQUAL_MESSAGE(CV_NO_TASK, trc,
"no continuation task was scheduled on the walAck-behind-target edge "
"(Req 5.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_stream_calls),
"the continuation must stream exactly once (Req 5.1)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(acked_lsn,
atomic_load(&g_stream_last_standby_lsn),
"the continuation must stream starting just above the ACKed LSN "
"(ha_stream_wal_to_standby streams (standby_lsn, target] = "
"[acked_lsn+1, target]) (Req 5.1)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(target,
atomic_load(&g_stream_last_target_lsn),
"the continuation must stream up to target = max(committed_lsn, "
"last_written) (Req 5.1)");
cv_teardown();
printf("test_walack_behind_target_schedules_one_continuation: checked\n");
}
/* 5.2 single-flight: a SECOND walAck while a continuation is already in flight
* for the same standby schedules NOTHING.
* RED: the scaffold schedules nothing for the first ACK either, so the
* "exactly one continuation across two ACKs" assertion fails (0 != 1). */
static void test_walack_single_flight_no_duplicate_continuation(void)
{
cv_setup("single", /*n_peers=*/3, /*n_standbys=*/1,
/*appended_lsn=*/9, /*n_vcpus=*/8);
atomic_store(&g_cv.persist.committed_lsn, 4);
const int peer_idx = 0;
int tasks_before = cv_task_count(&g_cv.delaymgr);
/* Act: first walAck (behind target) → schedules one continuation and marks
* the standby's catchup_inflight. */
(void)ha_walack_continuation_for_test(&g_cv, &g_cv_eps[peer_idx], peer_idx,
/*acked_lsn=*/3);
int tasks_after_first = cv_task_count(&g_cv.delaymgr);
TEST_ASSERT_EQUAL_INT_MESSAGE(tasks_before + 1, tasks_after_first,
"the first walAck behind target must schedule one continuation (Req 5.2 "
"setup)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
atomic_load(&g_cv.ha.catchup_inflight[peer_idx]),
"scheduling a continuation must set the per-standby single-flight flag "
"(Req 5.2)");
/* Act: a SECOND walAck while the continuation is still in flight (the task
* has NOT been drained). */
(void)ha_walack_continuation_for_test(&g_cv, &g_cv_eps[peer_idx], peer_idx,
/*acked_lsn=*/5);
/* Assert (Req 5.2): no duplicate continuation — task count unchanged. */
TEST_ASSERT_EQUAL_INT_MESSAGE(tasks_after_first,
cv_task_count(&g_cv.delaymgr),
"a second walAck while a continuation is in flight must schedule NOTHING "
"(single-flight, Req 5.2)");
/* Drain the single in-flight continuation (clears the flag). */
(void)cv_run_pending_task(&g_cv.delaymgr);
cv_teardown();
printf("test_walack_single_flight_no_duplicate_continuation: checked\n");
}
/* 5.2 release: after the in-flight continuation completes (flag cleared), the
* NEXT lagging walAck schedules a fresh continuation again.
* RED: the scaffold never schedules/clears, so neither continuation appears. */
static void test_walack_reschedules_after_continuation_completes(void)
{
cv_setup("respawn", /*n_peers=*/3, /*n_standbys=*/1,
/*appended_lsn=*/9, /*n_vcpus=*/8);
atomic_store(&g_cv.persist.committed_lsn, 4);
const int peer_idx = 0;
/* Act 1: first walAck → continuation #1. */
(void)ha_walack_continuation_for_test(&g_cv, &g_cv_eps[peer_idx], peer_idx,
/*acked_lsn=*/3);
int trc1 = cv_run_pending_task(&g_cv.delaymgr);
TEST_ASSERT_NOT_EQUAL_MESSAGE(CV_NO_TASK, trc1,
"first walAck must schedule a continuation (Req 5.2 release setup)");
/* After the continuation completes the single-flight flag must be clear. */
TEST_ASSERT_EQUAL_INT_MESSAGE(0,
atomic_load(&g_cv.ha.catchup_inflight[peer_idx]),
"the continuation must clear the single-flight flag on completion "
"(Req 5.2 release)");
int streams_after_first = atomic_load(&g_stream_calls);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, streams_after_first,
"the first continuation must have streamed once");
/* Act 2: a later lagging walAck (still behind target 9) → continuation #2. */
(void)ha_walack_continuation_for_test(&g_cv, &g_cv_eps[peer_idx], peer_idx,
/*acked_lsn=*/6);
int trc2 = cv_run_pending_task(&g_cv.delaymgr);
TEST_ASSERT_NOT_EQUAL_MESSAGE(CV_NO_TASK, trc2,
"a lagging walAck after the prior continuation completed must schedule "
"AGAIN (Req 5.2 release)");
TEST_ASSERT_EQUAL_INT_MESSAGE(2, atomic_load(&g_stream_calls),
"the second continuation must stream once more after the first released "
"(Req 5.2 release)");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(6,
atomic_load(&g_stream_last_standby_lsn),
"the second continuation streams from the newly ACKed LSN (Req 5.2 "
"release)");
cv_teardown();
printf("test_walack_reschedules_after_continuation_completes: checked\n");
}
/* 7.1 parity: a walAck with acked_lsn >= target calls repl_complete_catchup,
* which clears the standby's in_catch_up (re-inclusion in live replication).
* RED: the scaffold never calls repl_complete_catchup, so in_catch_up — set here
* via repl_start_catchup — stays 1. */
static void test_walack_at_parity_completes_catchup_and_clears_in_catch_up(void)
{
/* appended_lsn=9 → last_written=9; committed_lsn=9 → target=9. */
cv_setup("parity", /*n_peers=*/3, /*n_standbys=*/1,
/*appended_lsn=*/9, /*n_vcpus=*/8);
atomic_store(&g_cv.persist.committed_lsn, 9);
const int peer_idx = 0;
repl_ep_t ep = &g_cv_eps[peer_idx];
const uint64_t acked_lsn = 9; /* at parity with target 9 */
/* Put the standby into catch-up so re-inclusion is observable. */
repl_start_catchup(&g_cv.persist.repl, ep, /*join_applied_lsn=*/0);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, cv_standby_in_catch_up(ep),
"fixture precondition: the standby must be in_catch_up before parity ACK");
int tasks_before = cv_task_count(&g_cv.delaymgr);
/* Act: the standby acknowledges the target → parity. */
(void)ha_walack_continuation_for_test(&g_cv, ep, peer_idx, acked_lsn);
/* Assert (Req 7.1): no continuation scheduled at parity, and the standby is
* re-included — in_catch_up cleared by repl_complete_catchup. */
TEST_ASSERT_EQUAL_INT_MESSAGE(tasks_before, cv_task_count(&g_cv.delaymgr),
"a parity walAck must NOT schedule a streaming continuation (Req 7.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_stream_calls),
"a parity walAck must NOT stream (nothing left to send) (Req 7.1)");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, cv_standby_in_catch_up(ep),
"a parity walAck must call repl_complete_catchup, clearing the standby's "
"in_catch_up so it re-enters live replication (Req 7.1)");
cv_teardown();
printf("test_walack_at_parity_completes_catchup_and_clears_in_catch_up: checked\n");
}
/* 5.3 decision pin: the heartbeat decision function ha_evaluate_repl_action is
* the fallback initiator of catch-up and the SOLE owner of the snapshot
* decision; the ACK-driven continuation must not change its verdicts. This pins
* the existing boundary so task 4.2 cannot disturb the heartbeat path.
*
* This case exercises only existing pure-function behavior, so it PASSES against
* the current tree (intended — it locks the boundary; the RED signal for task
* 4.1 comes from the continuation/single-flight/parity cases above). */
static void test_heartbeat_decision_path_unchanged(void)
{
/* HA_REPL_STREAM_WAL: tag current, lsn within the WAL window
* (oldest_wal_lsn <= standby_lsn < committed_lsn). */
ha_repl_action_t a_stream = ha_evaluate_repl_action(
/*standby_tag=*/5, /*standby_lsn=*/3,
/*committed_tag=*/5, /*committed_lsn=*/9, /*oldest_wal_lsn=*/1);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)HA_REPL_STREAM_WAL, (int)a_stream,
"heartbeat decision: standby_lsn within the WAL window must remain "
"HA_REPL_STREAM_WAL — the heartbeat path stays the fallback catch-up "
"initiator (Req 5.3)");
/* HA_REPL_SEND_SNAPSHOT: standby_lsn predates the oldest retained WAL. */
ha_repl_action_t a_snap = ha_evaluate_repl_action(
/*standby_tag=*/5, /*standby_lsn=*/2,
/*committed_tag=*/5, /*committed_lsn=*/9, /*oldest_wal_lsn=*/5);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)HA_REPL_SEND_SNAPSHOT, (int)a_snap,
"heartbeat decision: standby_lsn below oldest_wal_lsn must remain "
"HA_REPL_SEND_SNAPSHOT — the heartbeat path remains the SOLE snapshot "
"decision owner (Req 5.3); the ACK-driven continuation never decides "
"snapshots");
printf("test_heartbeat_decision_path_unchanged: checked\n");
}
#endif /* XCE_TESTING */
/* ── Suite entry point ──────────────────────────────────────────────────── */
void test_convergence_verification(void)
{
#ifdef XCE_TESTING
printf("\n>> ACK-driven catch-up continuation tests (RED, task 4.1)\n\n");
RUN_TEST(test_walack_behind_target_schedules_one_continuation);
RUN_TEST(test_walack_single_flight_no_duplicate_continuation);
RUN_TEST(test_walack_reschedules_after_continuation_completes);
RUN_TEST(test_walack_at_parity_completes_catchup_and_clears_in_catch_up);
RUN_TEST(test_heartbeat_decision_path_unchanged);
#else
printf("test_convergence_verification: SKIP (non-test build)\n");
#endif
}