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

test(ha-convergence-recovery): regression pins for fan-out, snapshot, degraded-gate, re-inclusion (task 6.1)

Browse source
This commit is contained in:
Rongsong Shen 2026-06-14 06:56:57 +00:00
parent 07de4581b0
commit e399328339
2 changed files with 298 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

296
src/controller/testsuite/test_convergence_verification.c
View file

@ -55,6 +55,7 @@
#include <unity/unity.h>
#include "commit.h" /* commit_engine_t, commit_engine_init */
#include "controller.capnp.h" /* OPCode_* (task 6.1 (c) dispatch-gate pin) */
#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_* */
@ -63,6 +64,7 @@
#include "repl.h" /* repl_add_standby / repl_remove_standby /
* repl_active_standby_count / standby slots */
#include "xce/cluster_def.h" /* full cluster_t layout */
#include "xce/errnox.h" /* XCE_ERR_NOT_READY (task 6.1 (c) gate pin) */
#ifdef XCE_TESTING
@ -104,6 +106,37 @@ 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"
/* ── Fan-out counting send-hook (task 6.1 (a)/(d), umbrella 6.1/6.4 pins) ─────
* repl_broadcast routes each enqueue through repl_broadcast_send_hook under
* XCE_TESTING; this counting variant records, per standby slot, how many entries
* repl_broadcast tried to send to that slot's ep. An active, non-catch-up
* standby must receive exactly one send; an empty slot or an in_catch_up slot
* must receive ZERO. Indexed by matching ep against the g_cv_eps[] addresses.
* (Declared after g_cv_eps so the slot match resolves.) */
static atomic_int g_bcast_total_calls;
static atomic_int g_bcast_slot_calls[MAX_STANDBYS];
static int cv_count_send_hook(void *ep, wal_catchup_entry_t **entries,
int n_entries)
{
(void)entries; (void)n_entries;
atomic_fetch_add(&g_bcast_total_calls, 1);
for (int i = 0; i < MAX_STANDBYS; i++) {
if (ep == (void *)&g_cv_eps[i]) {
atomic_fetch_add(&g_bcast_slot_calls[i], 1);
break;
}
}
return 0;
}
static void cv_bcast_reset_counters(void)
{
atomic_store(&g_bcast_total_calls, 0);
for (int i = 0; i < MAX_STANDBYS; i++)
atomic_store(&g_bcast_slot_calls[i], 0);
}
static ha_config_t cv_ha_cfg(void)
{
ha_config_t cfg = {
@ -687,6 +720,261 @@ static void test_quorum_second_cycle_logs_again(void)
printf("test_quorum_second_cycle_logs_again: checked\n");
}
/* ════════════════════════════════════════════════════════════════════════════
* Task 6.1 — REGRESSION PINS (verification of existing behavior, NO production
* change). These encode behavior already present in the current tree; each pin
* PASSES against the code as-is. A failure here is a real regression, not a
* RED-phase placeholder.
*
* (a) repl_broadcast full fan-out — umbrella 6.1 pin (Req 6.1)
* (b) ha_evaluate_repl_action snapshot fallback boundary — umbrella 6.3 pin
* (Req 6.1, decision boundary only)
* (c) ha_dispatch_gate degraded-mode allowlist — umbrella 7.3 pin (Req 7.1)
* (d) repl_complete_catchup re-inclusion mechanism — umbrella 6.4 pin
* (Req 7.1, complements task 4's ACK-driven trigger test)
* ════════════════════════════════════════════════════════════════════════════ */
/* Drive one repl_broadcast of a single WAL_NOOP entry (no payload). */
static int cv_broadcast_one(replicator_t *r, uint64_t lsn)
{
wal_entry_hdr_t hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.lsn = lsn;
hdr.type = WAL_NOOP;
hdr.payload_len = 0;
return repl_broadcast(r, &hdr, NULL, 0);
}
/* (a) umbrella 6.1: repl_broadcast enqueues to EVERY registered active standby
* and skips ONLY empty slots and in_catch_up standbys.
*
* Fixture: three standby slots — slot 0 active, slot 1 emptied (removed), slot 2
* active but in_catch_up. A single broadcast must hit slot 0 exactly once and
* NEITHER slot 1 (empty) NOR slot 2 (catch-up). Pins existing repl.c:259-307. */
static void test_pin_repl_broadcast_full_fanout_skips_empty_and_catchup(void)
{
cv_setup("fanout", /*n_peers=*/3, /*n_standbys=*/3,
/*appended_lsn=*/0, /*n_vcpus=*/8);
/* Swap the drop hook for the counting hook and reset its tallies. */
cv_bcast_reset_counters();
repl_broadcast_send_hook = cv_count_send_hook;
/* Arrange the three slots into the three states under test. */
repl_remove_standby(&g_cv.persist.repl, &g_cv_eps[1]); /* slot 1 → empty */
repl_start_catchup(&g_cv.persist.repl, &g_cv_eps[2],
/*join_applied_lsn=*/0); /* slot 2 → catch-up */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, cv_standby_in_catch_up(&g_cv_eps[2]),
"fixture precondition: slot 2 must be in_catch_up before broadcast");
/* Act: one broadcast. */
int rc = cv_broadcast_one(&g_cv.persist.repl, /*lsn=*/1);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "repl_broadcast must succeed");
/* Assert (Req 6.1): the active non-catch-up standby got exactly one send. */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_bcast_slot_calls[0]),
"repl_broadcast must enqueue to the active non-catch-up standby (slot 0) "
"exactly once (umbrella 6.1 full fan-out)");
/* Assert: the EMPTY slot was skipped (no ep → no send). */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_bcast_slot_calls[1]),
"repl_broadcast must skip the empty slot (slot 1) (umbrella 6.1)");
/* Assert: the in_catch_up standby was skipped. */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_bcast_slot_calls[2]),
"repl_broadcast must skip the in_catch_up standby (slot 2) "
"(umbrella 6.1)");
/* Assert: exactly one send total — no over-fan-out. */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_bcast_total_calls),
"repl_broadcast must enqueue exactly once across all slots in this "
"fixture (one active, one empty, one catch-up) (umbrella 6.1)");
cv_teardown();
printf("test_pin_repl_broadcast_full_fanout_skips_empty_and_catchup: checked\n");
}
/* (b) umbrella 6.3: ha_evaluate_repl_action returns HA_REPL_SEND_SNAPSHOT when
* standby_lsn < oldest_wal_lsn (the standby predates the oldest retained WAL, so
* streaming cannot reach it — a fresh snapshot is required). Decision boundary
* only, per the task: at/above oldest_wal_lsn the verdict is NOT snapshot.
* Pins existing ha.c:1274-1275. (The snapshot send path itself —
* ha_send_snapshot_to_standby, config_repl.c:1857 — takes a fresh checkpoint
* before sending; that send-internals behavior is out of scope for this
* decision-boundary pin.) */
static void test_pin_evaluate_repl_action_snapshot_below_oldest_wal(void)
{
/* Below the oldest retained WAL → SEND_SNAPSHOT. tags equal (no config push
* shortcut); committed_lsn high so the only deciding factor is the snapshot
* boundary. */
ha_repl_action_t below = 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)below,
"standby_lsn (2) below oldest_wal_lsn (5) must decide SEND_SNAPSHOT "
"(umbrella 6.3 fallback boundary)");
/* Exactly AT oldest_wal_lsn is inside the retained window → NOT snapshot
* (streamable). The boundary is strict-less-than. */
ha_repl_action_t at_boundary = ha_evaluate_repl_action(
/*standby_tag=*/5, /*standby_lsn=*/5,
/*committed_tag=*/5, /*committed_lsn=*/9, /*oldest_wal_lsn=*/5);
TEST_ASSERT_NOT_EQUAL_MESSAGE((int)HA_REPL_SEND_SNAPSHOT, (int)at_boundary,
"standby_lsn == oldest_wal_lsn is within the retained WAL window — the "
"snapshot boundary is strict-less-than, so this must NOT decide "
"SEND_SNAPSHOT (umbrella 6.3 boundary)");
TEST_ASSERT_EQUAL_INT_MESSAGE((int)HA_REPL_STREAM_WAL, (int)at_boundary,
"at the snapshot boundary with lsn (5) below committed (9) the standby is "
"streamable — HA_REPL_STREAM_WAL (umbrella 6.3 boundary)");
/* Above the oldest WAL and at committed → CONVERGED, never snapshot. */
ha_repl_action_t converged = ha_evaluate_repl_action(
/*standby_tag=*/5, /*standby_lsn=*/9,
/*committed_tag=*/5, /*committed_lsn=*/9, /*oldest_wal_lsn=*/5);
TEST_ASSERT_NOT_EQUAL_MESSAGE((int)HA_REPL_SEND_SNAPSHOT, (int)converged,
"a fully caught-up standby must never decide SEND_SNAPSHOT "
"(umbrella 6.3 boundary)");
printf("test_pin_evaluate_repl_action_snapshot_below_oldest_wal: checked\n");
}
/* ── Degraded-mode dispatch-gate fixture (task 6.1 (c)) ──────────────────────
* Self-contained ha_state_t with role=CTRL_PRIMARY and write quorum ABSENT
* (n_peers=3, zero connected standby endpoints → ha_has_write_quorum false via
* the standby_endpoints fallback count). Mirrors test_commit_barrier_rollout.c's
* fact_gate_setup. No persist/repl needed — the gate reads role + quorum only. */
static ha_state_t g_cvg_ha;
static int g_cvg_ep; /* dummy endpoint (zarray stores its address) */
static void cv_gate_setup(int n_peers, int n_endpoints)
{
ha_config_t cfg = cv_ha_cfg();
TEST_ASSERT_EQUAL_INT(0, ha_init(&g_cvg_ha, &cfg, 0));
g_cvg_ha.n_peers = n_peers;
atomic_store_explicit(&g_cvg_ha.role, (int32_t)CTRL_PRIMARY,
memory_order_release);
g_cvg_ha.standby_endpoints = calloc(1, sizeof(zarray_t));
TEST_ASSERT_NOT_NULL(g_cvg_ha.standby_endpoints);
zarray_init(g_cvg_ha.standby_endpoints, 4, NULL);
for (int i = 0; i < n_endpoints; i++)
zarray_store(g_cvg_ha.standby_endpoints, &g_cvg_ep);
}
static void cv_gate_teardown(void)
{
if (g_cvg_ha.standby_endpoints) {
zarray_fini(g_cvg_ha.standby_endpoints);
free(g_cvg_ha.standby_endpoints);
g_cvg_ha.standby_endpoints = NULL;
}
ha_fini(&g_cvg_ha);
}
/* (c) umbrella 7.3: without write quorum, ha_dispatch_gate admits OPCode_get,
* heartbeat, the replication-opcode allowlist, and the fact-shaped opcodes
* (notify/update/setpreason/clrpreason — admitted per ha-commit-barrier-rollout's
* fact allowlist), while rejecting request-shaped opcodes
* (jobsub/jobcontrol/queuecontrol/hostcontrol/jobdispatch/join) with
* -XCE_ERR_NOT_READY. Pins existing ha.c:1516-1551. */
static void test_pin_dispatch_gate_degraded_allowlist(void)
{
/* Arrange: CTRL_PRIMARY, 3-peer cluster, zero standbys → quorum absent. */
cv_gate_setup(/*n_peers=*/3, /*n_endpoints=*/0);
/* ── ADMITTED (return 0) ───────────────────────────────────────────── */
static const struct { int opcode; const char *name; } admitted[] = {
/* read-only query + probe */
{ OPCode_get, "OPCode_get" },
{ OPCode_probeConfigTag, "OPCode_probeConfigTag" },
/* heartbeat */
{ OPCode_heartbeat, "OPCode_heartbeat" },
/* replication allowlist */
{ OPCode_replConfig, "OPCode_replConfig" },
{ OPCode_replConfigPush, "OPCode_replConfigPush" },
{ OPCode_upgradeConfigAck, "OPCode_upgradeConfigAck" },
{ OPCode_hello, "OPCode_hello" },
{ OPCode_walAck, "OPCode_walAck" },
{ OPCode_replEntry, "OPCode_replEntry" },
{ OPCode_replAck, "OPCode_replAck" },
{ OPCode_replCatchUpDone, "OPCode_replCatchUpDone" },
{ OPCode_walCatchupEntry, "OPCode_walCatchupEntry" },
{ OPCode_snapshotTransfer, "OPCode_snapshotTransfer" },
/* fact-shaped ops (admitted per the fact allowlist) */
{ OPCode_notify, "OPCode_notify" },
{ OPCode_update, "OPCode_update" },
{ OPCode_setpreason, "OPCode_setpreason" },
{ OPCode_clrpreason, "OPCode_clrpreason" },
};
for (size_t i = 0; i < sizeof(admitted) / sizeof(admitted[0]); i++) {
int rc = ha_dispatch_gate(&g_cvg_ha, admitted[i].opcode);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, admitted[i].name);
}
/* ── REJECTED with -XCE_ERR_NOT_READY (request-shaped ops only) ────── */
static const struct { int opcode; const char *name; } rejected[] = {
{ OPCode_jobsub, "OPCode_jobsub" },
{ OPCode_jobcontrol, "OPCode_jobcontrol" },
{ OPCode_queuecontrol, "OPCode_queuecontrol" },
{ OPCode_hostcontrol, "OPCode_hostcontrol" },
{ OPCode_jobdispatch, "OPCode_jobdispatch" },
{ OPCode_join, "OPCode_join" },
};
for (size_t i = 0; i < sizeof(rejected) / sizeof(rejected[0]); i++) {
int rc = ha_dispatch_gate(&g_cvg_ha, rejected[i].opcode);
TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_NOT_READY, rc, rejected[i].name);
}
cv_gate_teardown();
printf("test_pin_dispatch_gate_degraded_allowlist: checked\n");
}
/* (d) umbrella 6.4 mechanism: repl_complete_catchup clears in_catch_up and the
* standby re-enters repl_broadcast iteration (i.e. it is sent to again).
* Complements task 4's ACK-driven *trigger* test by pinning the *mechanism*.
* Pins existing repl.c:602-628 + the broadcast re-inclusion at repl.c:265. */
static void test_pin_repl_complete_catchup_reincludes_in_broadcast(void)
{
cv_setup("reinc", /*n_peers=*/3, /*n_standbys=*/1,
/*appended_lsn=*/0, /*n_vcpus=*/8);
cv_bcast_reset_counters();
repl_broadcast_send_hook = cv_count_send_hook;
repl_ep_t ep = &g_cv_eps[0];
/* Arrange: put the only standby into catch-up → excluded from broadcast. */
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");
/* While in catch-up, a broadcast must NOT reach it. */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, cv_broadcast_one(&g_cv.persist.repl, 1),
"repl_broadcast must succeed");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_bcast_slot_calls[0]),
"an in_catch_up standby must be skipped by repl_broadcast (umbrella 6.4 "
"precondition)");
/* Act: complete catch-up → re-inclusion. */
repl_complete_catchup(&g_cv.persist.repl, ep);
/* Assert: in_catch_up cleared. */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, cv_standby_in_catch_up(ep),
"repl_complete_catchup must clear in_catch_up (umbrella 6.4)");
/* Assert: the standby re-enters broadcast iteration — a fresh broadcast now
* reaches it. */
cv_bcast_reset_counters();
TEST_ASSERT_EQUAL_INT_MESSAGE(0, cv_broadcast_one(&g_cv.persist.repl, 2),
"repl_broadcast must succeed");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_bcast_slot_calls[0]),
"after repl_complete_catchup the standby must re-enter repl_broadcast "
"iteration and receive the next entry (umbrella 6.4 re-inclusion)");
cv_teardown();
printf("test_pin_repl_complete_catchup_reincludes_in_broadcast: checked\n");
}
#endif /* XCE_TESTING */
/* ── Suite entry point ──────────────────────────────────────────────────── */
@ -708,6 +996,14 @@ void test_convergence_verification(void)
RUN_TEST(test_quorum_loss_no_repeat_log_while_lost);
RUN_TEST(test_quorum_restore_logged_once);
RUN_TEST(test_quorum_second_cycle_logs_again);
printf("\n>> Regression pins: fan-out / snapshot fallback / degraded gate / "
"re-inclusion (task 6.1 — PASS against current code)\n\n");
RUN_TEST(test_pin_repl_broadcast_full_fanout_skips_empty_and_catchup);
RUN_TEST(test_pin_evaluate_repl_action_snapshot_below_oldest_wal);
RUN_TEST(test_pin_dispatch_gate_degraded_allowlist);
RUN_TEST(test_pin_repl_complete_catchup_reincludes_in_broadcast);
#else
printf("test_convergence_verification: SKIP (non-test build)\n");
#endif