From 052c891ac04dfc4d3d6df833f564cb2e75e290fa Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 30 May 2026 13:42:41 +0000 Subject: [PATCH] =?UTF-8?q?test(election-iomgr-refactor):=20add=20ha=5Frep?= =?UTF-8?q?l=5Fack=5Freceived=20quorum/promotion=20unit=20tests=20?= =?UTF-8?q?=E2=80=94=20task=206.2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + src/controller/testsuite/test_ha_repl_ack.c | 302 ++++++++++++++++++++ 3 files changed, 305 insertions(+) create mode 100644 src/controller/testsuite/test_ha_repl_ack.c diff --git a/src/controller/meson.build b/src/controller/meson.build index 93e448f9..f5f69775 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -116,6 +116,7 @@ controller_unit_test_src = [ 'testsuite/test_ha_vote_gate.c', 'testsuite/test_ha_evaluate_committing.c', 'testsuite/test_ha_repl_push.c', + 'testsuite/test_ha_repl_ack.c', 'startup_validation.c', 'worker.c', 'persist.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 6da04096..0c6ec1df 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -48,6 +48,7 @@ extern void test_ha_start_election_async(void); extern void test_ha_vote_gate(void); extern void test_ha_evaluate_committing(void); extern void test_ha_repl_push(void); +extern void test_ha_repl_ack(void); void setUp(void) { return; } @@ -165,6 +166,7 @@ int main(int argc, char* argv[]) TEST(test_ha_vote_gate); TEST(test_ha_evaluate_committing); TEST(test_ha_repl_push); + TEST(test_ha_repl_ack); return UNITY_END(); } diff --git a/src/controller/testsuite/test_ha_repl_ack.c b/src/controller/testsuite/test_ha_repl_ack.c new file mode 100644 index 00000000..93da015e --- /dev/null +++ b/src/controller/testsuite/test_ha_repl_ack.c @@ -0,0 +1,302 @@ +/* test_ha_repl_ack.c — Unity tests for Task 6.2: ha_repl_ack_received quorum logic. + * + * Tests verify (Req 8.2, 8.3): + * Req 8.2: When acks reach quorum-1 threshold, ha_promote_to_primary is called. + * Req 8.3: CTRL_PRIMARY opens full traffic (dispatch gate PRIMARY case). + * + * Quorum formula: quorum = (n_peers + 2) / 2 + * n_peers=1 (solo) → quorum=1 → threshold=0 (quorum-1=0) → first ack promotes + * n_peers=2 (3-node)→ quorum=2 → threshold=1 → first ack promotes + * n_peers=3 (4-node)→ quorum=2 → threshold=1 → first ack promotes + * n_peers=4 (5-node)→ quorum=3 → threshold=2 → second ack promotes (acks >= quorum-1) + * + * Boundary: ha.c unit tests (ha_repl_ack_received, ha_dispatch_gate) + * Requirements: 8.2, 8.3 + */ + +#define _GNU_SOURCE +#include +#include +#include + +#include + +#include "ha.h" +#include "xce/cluster_def.h" + +/* ha_repl_ack_received declaration (defined in ha.c) */ +void ha_repl_ack_received(ha_state_t *ha, cluster_t *c); + +/* ha_dispatch_gate declaration (defined in ha.c) */ +int ha_dispatch_gate(ha_state_t *ha, int opcode); + +/* OPCode constants needed */ +#include "controller.capnp.h" + +/* ── Shared HA config ──────────────────────────────────────────────────── */ + +static ha_config_t g_ack_cfg = { + .heartbeat_interval_ms = 3000, + .heartbeat_timeout_ms = 10000, + .election_timeout_ms = 2000, + .probe_connect_timeout_ms = 500, + .max_promotion_retries = 5, + .election_jitter_max_ms = 0, + .election_priority_step_ms = 500, + .committing_timeout_ms = 10000, +}; + +/* ── Helper: build a minimally-zeroed cluster_t with ha_init applied ─── */ + +static void init_cluster_committing(cluster_t *c, int n_peers) +{ + memset(c, 0, sizeof(*c)); + ha_init(&c->ha, &g_ack_cfg, 0); + c->ha.n_peers = n_peers; + /* Simulate being in COMMITTING state (election won, waiting for acks) */ + atomic_store_explicit(&c->ha.role, (int32_t)CTRL_PRIMARY_COMMITTING, + memory_order_release); + atomic_store_explicit(&c->ha.repl_ack_count, 0, memory_order_release); +} + +/* ── Req 8.2: 2-node cluster (n_peers=2, quorum=2, threshold=1) ──────── */ +/* With 2 peers (3 total nodes), quorum=2, threshold=quorum-1=1. + * First ack reaches threshold → promote. */ + +static void test_repl_ack_2node_first_ack_promotes(void) +{ + /* n_peers=2 means 3 nodes total (self + 2 peers). + * quorum = (2+2)/2 = 2; threshold = quorum-1 = 1. + * First ack (count becomes 1 >= 1) → ha_promote_to_primary called. */ + cluster_t c; + init_cluster_committing(&c, 2); + + ha_repl_ack_received(&c.ha, &c); + + int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role, + "2-node: first ack (count=1 >= threshold=1) must promote to CTRL_PRIMARY (Req 8.2)"); + + ha_fini(&c.ha); + printf("test_repl_ack_2node_first_ack_promotes: PASS\n"); +} + +/* ── Req 8.2: 2-node cluster — 0 acks → stays COMMITTING ────────────── */ + +static void test_repl_ack_2node_no_ack_stays_committing(void) +{ + /* With 0 acks, role must remain CTRL_PRIMARY_COMMITTING. */ + cluster_t c; + init_cluster_committing(&c, 2); + + /* Do NOT call ha_repl_ack_received — verify initial state */ + int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role, + "2-node: 0 acks → role must remain CTRL_PRIMARY_COMMITTING (Req 8.2)"); + + ha_fini(&c.ha); + printf("test_repl_ack_2node_no_ack_stays_committing: PASS\n"); +} + +/* ── Req 8.2: 4-node cluster (n_peers=3, quorum=2, threshold=1) ──────── */ +/* quorum = (3+2)/2 = 2; threshold=1. First ack promotes. */ + +static void test_repl_ack_4node_first_ack_promotes(void) +{ + cluster_t c; + init_cluster_committing(&c, 3); + + ha_repl_ack_received(&c.ha, &c); + + int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role, + "4-node: first ack (count=1 >= threshold=1) must promote to CTRL_PRIMARY (Req 8.2)"); + + ha_fini(&c.ha); + printf("test_repl_ack_4node_first_ack_promotes: PASS\n"); +} + +/* ── Req 8.2: 5-node cluster (n_peers=4, quorum=3, threshold=2) ──────── */ +/* quorum = (4+2)/2 = 3; threshold=2. + * First ack (count=1) below threshold → stays COMMITTING. + * Second ack (count=2) reaches threshold → promotes. */ + +static void test_repl_ack_5node_first_ack_stays_committing(void) +{ + cluster_t c; + init_cluster_committing(&c, 4); + + ha_repl_ack_received(&c.ha, &c); /* ack count → 1, threshold=2 → no promote */ + + int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role, + "5-node: first ack (count=1 < threshold=2) must stay COMMITTING (Req 8.2)"); + + ha_fini(&c.ha); + printf("test_repl_ack_5node_first_ack_stays_committing: PASS\n"); +} + +static void test_repl_ack_5node_second_ack_promotes(void) +{ + cluster_t c; + init_cluster_committing(&c, 4); + + ha_repl_ack_received(&c.ha, &c); /* ack count → 1 */ + /* Reset role back to COMMITTING so we can test from COMMITTING again. + * (In production ha_promote_to_primary isn't called twice; here we need + * to test the threshold on the second call.) */ + /* We start clean from count=1 and add one more via a second call. */ + + ha_fini(&c.ha); + + /* Fresh cluster with pre-populated ack_count=1 to test the crossing point */ + cluster_t c2; + init_cluster_committing(&c2, 4); + atomic_store_explicit(&c2.ha.repl_ack_count, 1, memory_order_release); + + ha_repl_ack_received(&c2.ha, &c2); /* ack count → 2 >= threshold=2 → promote */ + + int role = (int)atomic_load_explicit(&c2.ha.role, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role, + "5-node: second ack (count=2 >= threshold=2) must promote to CTRL_PRIMARY (Req 8.2)"); + + ha_fini(&c2.ha); + printf("test_repl_ack_5node_second_ack_promotes: PASS\n"); +} + +/* ── Req 8.2: null cluster_t — ha_repl_ack_received must not crash ───── */ +/* Design doc: ha_repl_ack_received(ha, c=NULL) increments ack but skips promote. */ + +static void test_repl_ack_null_cluster_increments_count(void) +{ + ha_state_t ha; + ha_init(&ha, &g_ack_cfg, 0); + ha.n_peers = 2; + atomic_store_explicit(&ha.role, (int32_t)CTRL_PRIMARY_COMMITTING, + memory_order_release); + atomic_store_explicit(&ha.repl_ack_count, 0, memory_order_release); + + /* c=NULL: must not crash; quorum check still fires but ha_promote_to_primary + * is guarded by `if (c)` in ha_repl_ack_received. */ + ha_repl_ack_received(&ha, NULL); + + int acks = atomic_load_explicit(&ha.repl_ack_count, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, acks, + "repl_ack with NULL cluster must still increment repl_ack_count (Req 8.2)"); + + /* Role should NOT have changed to PRIMARY (no cluster to promote via) */ + int role = (int)atomic_load_explicit(&ha.role, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role, + "repl_ack with NULL cluster must NOT promote role (no cluster_t available)"); + + ha_fini(&ha); + printf("test_repl_ack_null_cluster_increments_count: PASS\n"); +} + +/* ── Req 8.3: after promotion, dispatch gate allows full traffic ──────── */ +/* Once ha_repl_ack_received fires ha_promote_to_primary, role = CTRL_PRIMARY. + * ha_dispatch_gate in CTRL_PRIMARY state must allow job submission. */ + +static void test_repl_ack_after_promote_gate_allows_jobsub(void) +{ + cluster_t c; + init_cluster_committing(&c, 2); + + /* Trigger promotion via ack */ + ha_repl_ack_received(&c.ha, &c); + + /* Verify role is now PRIMARY */ + int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role, + "after ack promote: role must be CTRL_PRIMARY (Req 8.2)"); + + /* Dispatch gate in CTRL_PRIMARY must allow OPCode_jobsub (Req 8.3) */ + int rc = ha_dispatch_gate(&c.ha, OPCode_jobsub); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "CTRL_PRIMARY dispatch gate must allow OPCode_jobsub (Req 8.3)"); + + ha_fini(&c.ha); + printf("test_repl_ack_after_promote_gate_allows_jobsub: PASS\n"); +} + +/* ── Req 8.3: before promotion, gate still blocks jobsub ─────────────── */ +/* In COMMITTING state, before threshold is reached, job ops are blocked. */ + +static void test_repl_ack_committing_gate_blocks_jobsub_until_promote(void) +{ + /* 5-node cluster: threshold=2. First ack leaves us in COMMITTING. */ + cluster_t c; + init_cluster_committing(&c, 4); + + ha_repl_ack_received(&c.ha, &c); /* count=1, threshold=2 → still COMMITTING */ + + int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role, + "5-node after 1 ack: must remain COMMITTING before threshold (Req 8.2)"); + + /* Gate must block job submission while still COMMITTING */ + int rc = ha_dispatch_gate(&c.ha, OPCode_jobsub); + TEST_ASSERT_NOT_EQUAL_MESSAGE(0, rc, + "COMMITTING gate must block OPCode_jobsub (Req 8.3, 5.1)"); + + ha_fini(&c.ha); + printf("test_repl_ack_committing_gate_blocks_jobsub_until_promote: PASS\n"); +} + +/* ── Req 8.2: repl_ack_count accumulates correctly across calls ──────── */ + +static void test_repl_ack_count_increments_monotonically(void) +{ + ha_state_t ha; + ha_init(&ha, &g_ack_cfg, 0); + ha.n_peers = 10; /* large cluster so quorum > 3 calls */ + /* n_peers=10 → quorum=(10+2)/2=6 → threshold=5 */ + atomic_store_explicit(&ha.role, (int32_t)CTRL_PRIMARY_COMMITTING, + memory_order_release); + atomic_store_explicit(&ha.repl_ack_count, 0, memory_order_release); + + ha_repl_ack_received(&ha, NULL); /* count → 1 */ + ha_repl_ack_received(&ha, NULL); /* count → 2 */ + ha_repl_ack_received(&ha, NULL); /* count → 3 */ + + int acks = atomic_load_explicit(&ha.repl_ack_count, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE(3, acks, + "repl_ack_count must monotonically increment with each ack call (Req 8.2)"); + + ha_fini(&ha); + printf("test_repl_ack_count_increments_monotonically: PASS\n"); +} + +/* ── Req 8.2: null ha pointer — must not crash ────────────────────────── */ + +static void test_repl_ack_null_ha_is_noop(void) +{ + /* ha_repl_ack_received(NULL, ...) must return without crashing */ + ha_repl_ack_received(NULL, NULL); + printf("test_repl_ack_null_ha_is_noop: PASS\n"); +} + +/* ── Test suite entry point ───────────────────────────────────────────── */ + +void test_ha_repl_ack(void) +{ + /* Req 8.2 — quorum threshold: 2-node (3 total nodes) */ + RUN_TEST(test_repl_ack_2node_first_ack_promotes); + RUN_TEST(test_repl_ack_2node_no_ack_stays_committing); + + /* Req 8.2 — quorum threshold: 4-node (n_peers=3) */ + RUN_TEST(test_repl_ack_4node_first_ack_promotes); + + /* Req 8.2 — quorum threshold: 5-node (n_peers=4, threshold=2) */ + RUN_TEST(test_repl_ack_5node_first_ack_stays_committing); + RUN_TEST(test_repl_ack_5node_second_ack_promotes); + + /* Req 8.2 — null cluster guard and ack count accumulation */ + RUN_TEST(test_repl_ack_null_cluster_increments_count); + RUN_TEST(test_repl_ack_count_increments_monotonically); + RUN_TEST(test_repl_ack_null_ha_is_noop); + + /* Req 8.3 — dispatch gate reflects promotion state */ + RUN_TEST(test_repl_ack_after_promote_gate_allows_jobsub); + RUN_TEST(test_repl_ack_committing_gate_blocks_jobsub_until_promote); +}