From 757558bb33e6712d4bb0eede68e5c482f2d8c040 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 30 May 2026 11:56:31 +0000 Subject: [PATCH] =?UTF-8?q?feat(election-iomgr-refactor):=20config/WAL=20r?= =?UTF-8?q?eplication=20push=20to=20standbys=20in=20COMMITTING=20path=20?= =?UTF-8?q?=E2=80=94=20task=204.2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/controller/cluster.c | 54 ++++ src/controller/ha.c | 41 +++ src/controller/ha.h | 24 ++ src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + src/controller/testsuite/test_ha_repl_push.c | 311 +++++++++++++++++++ 6 files changed, 433 insertions(+) create mode 100644 src/controller/testsuite/test_ha_repl_push.c diff --git a/src/controller/cluster.c b/src/controller/cluster.c index fb0891ce..3de63419 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -1229,6 +1229,60 @@ int cluster_run(cluster_t* c) return 0; } +/* xce_request: defined in request.c, not yet in any public header */ +extern int xce_request(endpoint_t *ep, xce_request_t *req); + +/* ── Task 4.2: ha_send_ctrlhello_to_all_standbys ─────────────────────────── + * + * Send OP_hello CtrlHelloRequest to every connected standby endpoint. + * This triggers handle_ctrlhello on the receiving standby, which runs the + * existing catch-up replication sequence (WAL streaming, config push). + * + * Must be called from the COMMITTING path in ha_elect_evaluate_and_signal + * after ha->role is set to CTRL_PRIMARY_COMMITTING (Req 7.2). + * + * Thread safety: acquires ha->lock around standby_endpoints iteration. + * Best-effort: failures per-endpoint are logged but do not abort the loop. + */ +void ha_send_ctrlhello_to_all_standbys(cluster_t *c) +{ + if (!c) return; + ha_state_t *ha = &c->ha; + if (!ha->standby_endpoints) return; + + uint64_t my_lsn = atomic_load_explicit(&c->persist.committed_lsn, + memory_order_acquire); + uint64_t my_tag = c->myconf ? c->myconf->config_version_tag : 0; + char *my_name = c->myconf ? c->myconf->name : ""; + + pthread_mutex_lock(&ha->lock); + int n = ha->standby_endpoints->count; + for (int i = 0; i < n; i++) { + endpoint_t *ep = (endpoint_t *)zarray_get(ha->standby_endpoints, i); + if (!ep) continue; + + ctrl_hello_req_t hello = { + .name = my_name, + .key = NULL, + .applied_lsn = my_lsn, + .config_tag = my_tag, + }; + + xce_request_t req; + memset(&req, 0, sizeof(req)); + req.opcode = OPCode_hello; + req.body_which = Request_body_ctrlhellomsg; + req.body.ctrlhellomsg = &hello; + + int rc = xce_request(ep, &req); + if (rc < 0) { + WARN("ha_send_ctrlhello_to_all_standbys: xce_request to standby[%d] " + "failed: %d", i, rc); + } + } + pthread_mutex_unlock(&ha->lock); +} + void cluster_fini(cluster_t* c) { /* Free heap-allocated standby_endpoints array (created in phase1) */ diff --git a/src/controller/ha.c b/src/controller/ha.c index 0b323e93..e9390b18 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -956,6 +956,15 @@ static int committing_timeout_fn(void *arg); * unit-test binary). */ #ifndef XCE_TESTING extern int cluster_init_phase2(cluster_t *c); +/* ha_send_ctrlhello_to_all_standbys: defined in cluster.c. + * Iterates ha->standby_endpoints, sends CtrlHelloRequest to each, triggering + * catch-up replication in handle_ctrlhello (Req 7.2). */ +extern void ha_send_ctrlhello_to_all_standbys(cluster_t *c); +/* push_config_to_standbys: defined in config_repl.c (Req 7.1). + * Best-effort fan-out of config blob to all connected standbys. + * Called with (c, NULL, 0, NULL, 0) to push from the primary's saved config. */ +extern int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, + const uint8_t sha256[32], uint64_t new_tag); #endif #ifdef XCE_TESTING @@ -965,6 +974,20 @@ extern int cluster_init_phase2(cluster_t *c); * cluster_init_phase2() so tests can count phase2 calls without a live cluster_t. */ int (*ha_phase2_hook)(void *c) = NULL; + +/* + * ha_send_ctrlhello_hook: test-only seam for ha_send_ctrlhello_to_all_standbys. + * When non-NULL, the COMMITTING path calls this INSTEAD OF the real function + * (cluster.c is not linked in unit-test binary). Req 7.2. + */ +int (*ha_send_ctrlhello_hook)(void *c) = NULL; + +/* + * ha_push_config_hook: test-only seam for push_config_to_standbys. + * When non-NULL, the COMMITTING path calls this INSTEAD OF the real function + * (config_repl.c's push_config_to_standbys needs a real cluster). Req 7.1. + */ +int (*ha_push_config_hook)(void *c) = NULL; #endif /* XCE_TESTING */ /* ── Election Evaluation (Req 2.1, 2.7, 2.8) ─────────────────────────────── @@ -1101,6 +1124,24 @@ apply_role: /* Reset ack counter for this committing round (Req 5.2) */ atomic_store(&ha->repl_ack_count, 0); + /* Push config to standbys with lower config_tag (Req 7.1) */ +#ifdef XCE_TESTING + if (ha_push_config_hook != NULL) + ha_push_config_hook(c); +#else + if (c != NULL) + push_config_to_standbys(c, NULL, 0, NULL, 0); +#endif + + /* Send CtrlHelloRequest to all standbys to trigger WAL catch-up (Req 7.2) */ +#ifdef XCE_TESTING + if (ha_send_ctrlhello_hook != NULL) + ha_send_ctrlhello_hook(c); +#else + if (c != NULL) + ha_send_ctrlhello_to_all_standbys(c); +#endif + /* Arm committing timeout watchdog (degraded-mode fallback, Req 8.4) */ if (c) { committing_timeout_ctx_t *ct = calloc(1, sizeof(*ct)); diff --git a/src/controller/ha.h b/src/controller/ha.h index f7d0c1e4..5d672393 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -507,6 +507,15 @@ int ha_find_and_connect_to_primary(ha_state_t *ha, */ void ha_repl_ack_received(ha_state_t *ha, cluster_t *c); +/* + * ha_send_ctrlhello_to_all_standbys: iterate ha->standby_endpoints and send + * an OP_hello CtrlHelloRequest to each connected standby. + * This triggers the existing catch-up replication path in handle_ctrlhello + * (Req 7.2). Defined in cluster.c; called from ha_elect_evaluate_and_signal + * in the COMMITTING path. + */ +void ha_send_ctrlhello_to_all_standbys(cluster_t *c); + /* ── P3-6: Test-only failover shim ────────────────────────────── */ #ifdef XCE_TESTING @@ -518,6 +527,21 @@ void ha_repl_ack_received(ha_state_t *ha, cluster_t *c); */ extern int (*ha_phase2_hook)(void *c); +/* + * ha_send_ctrlhello_hook: test-only seam for ha_send_ctrlhello_to_all_standbys. + * When non-NULL, ha_elect_evaluate_and_signal() calls this in the COMMITTING + * path instead of the real function (cluster.c not linked in test binary). + * Requirements: 7.2. + */ +extern int (*ha_send_ctrlhello_hook)(void *c); + +/* + * ha_push_config_hook: test-only seam for push_config_to_standbys. + * When non-NULL, ha_elect_evaluate_and_signal() calls this in the COMMITTING + * path instead of the real function. Requirements: 7.1. + */ +extern int (*ha_push_config_hook)(void *c); + /* Test-only shim: demote then re-promote; returns new term. * Exercises the manual failover state machine without a live cluster_t. * Calls ha_do_demote() then ha_do_promote(); validates role transitions diff --git a/src/controller/meson.build b/src/controller/meson.build index 48c46089..93e448f9 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -115,6 +115,7 @@ controller_unit_test_src = [ 'testsuite/test_ha_start_election_async.c', 'testsuite/test_ha_vote_gate.c', 'testsuite/test_ha_evaluate_committing.c', + 'testsuite/test_ha_repl_push.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 e5156ed2..6da04096 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -47,6 +47,7 @@ extern void test_ha_vote_send(void); 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); void setUp(void) { return; } @@ -163,6 +164,7 @@ int main(int argc, char* argv[]) TEST(test_ha_start_election_async); TEST(test_ha_vote_gate); TEST(test_ha_evaluate_committing); + TEST(test_ha_repl_push); return UNITY_END(); } diff --git a/src/controller/testsuite/test_ha_repl_push.c b/src/controller/testsuite/test_ha_repl_push.c new file mode 100644 index 00000000..22b807c7 --- /dev/null +++ b/src/controller/testsuite/test_ha_repl_push.c @@ -0,0 +1,311 @@ +/* + * test_ha_repl_push.c — Unity tests for Task 4.2: + * Config and WAL replication push to standbys in the COMMITTING path. + * + * Tests verify: + * Req 7.1: When entering CTRL_PRIMARY_COMMITTING, push_config_to_standbys + * is called (via ha_push_config_hook seam). + * Req 7.2: When entering CTRL_PRIMARY_COMMITTING, ha_send_ctrlhello_to_all_standbys + * is called to trigger catch-up replication (via ha_send_ctrlhello_hook seam). + * Req 7.3: The push is triggered at the point of COMMITTING entry, not before. + * When election is lost (CTRL_STANDBY), no push occurs. + * + * Design ref: design.md "ha_send_ctrlhello_to_all_standbys" section + * Boundary: ha.c (election evaluation + COMMITTING path) + * Requirements: 7.1, 7.2, 7.3 + */ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "ha.h" + +/* ── Test-only seam: ha_send_ctrlhello_to_all_standbys call hook ───────────── + * + * In production, ha_elect_evaluate_and_signal calls + * ha_send_ctrlhello_to_all_standbys(c) in the COMMITTING path. + * For unit tests, the real function is not linked (cluster.c not in test binary). + * The XCE_TESTING seam in ha.c checks ha_send_ctrlhello_hook; if non-NULL, it + * calls the hook instead of the real ha_send_ctrlhello_to_all_standbys. + */ +extern int (*ha_send_ctrlhello_hook)(void *c); + +/* ── Test-only seam: push_config_to_standbys call hook ────────────────────── + * + * In production, ha_elect_evaluate_and_signal calls push_config_to_standbys(c,...) + * in the COMMITTING path. For unit tests, the seam in ha.c checks + * ha_push_config_hook; if non-NULL, it calls the hook instead. + */ +extern int (*ha_push_config_hook)(void *c); + +/* ── Call counters for hooks ─────────────────────────────────────────────── */ +static int g_ctrlhello_call_count = 0; +static int g_push_config_call_count = 0; + +static int stub_ha_send_ctrlhello(void *c) +{ + (void)c; + g_ctrlhello_call_count++; + return 0; +} + +static int stub_ha_push_config(void *c) +{ + (void)c; + g_push_config_call_count++; + return 0; +} + +/* Also need the phase2 hook since ha_elect_evaluate_and_signal calls it */ +extern int (*ha_phase2_hook)(void *c); + +static int stub_phase2(void *c) +{ + (void)c; + return 0; +} + +/* ── Minimal ha_config_t for tests ─────────────────────────────────────── */ + +static ha_config_t g_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 minimal ha_elect_ctx_t ─────────────────────────────── */ + +static ha_elect_ctx_t *make_ctx(int n_other, int votes_granted) +{ + ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) + + (size_t)n_other * sizeof(ctx->votes[0])); + if (!ctx) return NULL; + ctx->n_peers = n_other; + ctx->votes_granted = votes_granted; + ctx->n_replied = n_other; + ctx->max_resp_tag = 0; + ctx->max_resp_lsn = 0; + ctx->result = -1; + ctx->done_posted = 0; + ctx->c = NULL; /* no real cluster in unit tests */ + pthread_mutex_init(&ctx->lock, NULL); + sem_init(&ctx->done, 0, 0); + if (n_other > 0) { + ctx->votes[0].my_tag = 100; + ctx->votes[0].my_lsn = 50; + } + return ctx; +} + +static void free_ctx(ha_elect_ctx_t *ctx) +{ + if (!ctx) return; + pthread_mutex_destroy(&ctx->lock); + sem_destroy(&ctx->done); + free(ctx); +} + +/* ── Setup / teardown ────────────────────────────────────────────────────── */ + +static void reset_counters(void) +{ + g_ctrlhello_call_count = 0; + g_push_config_call_count = 0; + ha_send_ctrlhello_hook = stub_ha_send_ctrlhello; + ha_push_config_hook = stub_ha_push_config; + ha_phase2_hook = stub_phase2; +} + +static void clear_hooks(void) +{ + ha_send_ctrlhello_hook = NULL; + ha_push_config_hook = NULL; + ha_phase2_hook = NULL; +} + +/* ── Test 1: Req 7.2 — COMMITTING path calls ha_send_ctrlhello_to_all_standbys ── */ + +static void test_committing_calls_ctrlhello_push(void) +{ + /* When election is won (CTRL_PRIMARY → COMMITTING), ha_send_ctrlhello_to_all_standbys + * must be called exactly once. + * Single-node wins: n_other=0, votes_granted=1. */ + reset_counters(); + + ha_state_t ha; + ha_init(&ha, &g_cfg, 0); + ha.phase2_done = 1; /* skip phase2 */ + atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release); + + ha_elect_ctx_t *ctx = make_ctx(0, 1); /* single-node wins */ + TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "make_ctx failed"); + + ha_elect_evaluate_and_signal(&ha, ctx); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_ctrlhello_call_count, + "ha_send_ctrlhello_to_all_standbys must be called once on COMMITTING entry (Req 7.2)"); + + free_ctx(ctx); + ha_fini(&ha); + clear_hooks(); + printf("test_committing_calls_ctrlhello_push: PASS (ctrlhello_calls=%d)\n", + g_ctrlhello_call_count); +} + +/* ── Test 2: Req 7.1 — COMMITTING path calls push_config_to_standbys ──────── */ + +static void test_committing_calls_push_config(void) +{ + /* When election is won, push_config_to_standbys must be called once. */ + reset_counters(); + + ha_state_t ha; + ha_init(&ha, &g_cfg, 0); + ha.phase2_done = 1; + atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release); + + ha_elect_ctx_t *ctx = make_ctx(0, 1); + TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "make_ctx failed"); + + ha_elect_evaluate_and_signal(&ha, ctx); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_push_config_call_count, + "push_config_to_standbys must be called once on COMMITTING entry (Req 7.1)"); + + free_ctx(ctx); + ha_fini(&ha); + clear_hooks(); + printf("test_committing_calls_push_config: PASS (push_config_calls=%d)\n", + g_push_config_call_count); +} + +/* ── Test 3: Req 7.3 — losing election → no push ───────────────────────── */ + +static void test_standby_result_does_not_push(void) +{ + /* When the election is lost (CTRL_STANDBY), neither push function is called. */ + reset_counters(); + + ha_state_t ha; + ha_init(&ha, &g_cfg, 0); + ha.phase2_done = 1; + atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release); + + /* n_other=2, quorum=2, votes_granted=1 → loses */ + ha_elect_ctx_t *ctx = make_ctx(2, 1); + TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "make_ctx failed"); + ctx->votes[0].my_tag = 100; + ctx->votes[0].my_lsn = 50; + + ctrl_role_t result = ha_elect_evaluate_and_signal(&ha, ctx); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_STANDBY, (int)result, + "election with votes_granted=1 of n_other=2 must lose"); + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_ctrlhello_call_count, + "ha_send_ctrlhello_to_all_standbys must NOT be called when standby (Req 7.3)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_push_config_call_count, + "push_config_to_standbys must NOT be called when standby (Req 7.3)"); + + free_ctx(ctx); + ha_fini(&ha); + clear_hooks(); + printf("test_standby_result_does_not_push: PASS " + "(ctrlhello=%d, push_config=%d)\n", + g_ctrlhello_call_count, g_push_config_call_count); +} + +/* ── Test 4: 3-node win also triggers both pushes ───────────────────────── */ + +static void test_3node_win_calls_both_pushes(void) +{ + /* n_other=2, quorum=2, votes_granted=3 → wins. + * Both push hooks must be called exactly once. */ + reset_counters(); + + ha_state_t ha; + ha_init(&ha, &g_cfg, 0); + ha.phase2_done = 1; + atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release); + + ha_elect_ctx_t *ctx = make_ctx(2, 3); /* votes_granted=3 (self + 2 peers) */ + TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "make_ctx failed"); + ctx->votes[0].my_tag = 100; + ctx->votes[0].my_lsn = 50; + + ctrl_role_t result = ha_elect_evaluate_and_signal(&ha, ctx); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, (int)result, + "3-node win must return CTRL_PRIMARY"); + + ctrl_role_t actual_role = + (ctrl_role_t)atomic_load_explicit(&ha.role, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, (int)actual_role, + "ha->role must be CTRL_PRIMARY_COMMITTING after winning"); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_ctrlhello_call_count, + "ha_send_ctrlhello_to_all_standbys must be called once (3-node win)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_push_config_call_count, + "push_config_to_standbys must be called once (3-node win)"); + + free_ctx(ctx); + ha_fini(&ha); + clear_hooks(); + printf("test_3node_win_calls_both_pushes: PASS " + "(ctrlhello=%d, push_config=%d)\n", + g_ctrlhello_call_count, g_push_config_call_count); +} + +/* ── Test 5: done_posted guard prevents double push ─────────────────────── */ + +static void test_done_posted_prevents_double_push(void) +{ + /* Second call to ha_elect_evaluate_and_signal on same ctx is a no-op + * (done_posted guard): neither push hook should fire a second time. */ + reset_counters(); + + ha_state_t ha; + ha_init(&ha, &g_cfg, 0); + ha.phase2_done = 1; + atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release); + + ha_elect_ctx_t *ctx = make_ctx(0, 1); + TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "make_ctx failed"); + + ha_elect_evaluate_and_signal(&ha, ctx); /* first: triggers pushes (count=1) */ + ha_elect_evaluate_and_signal(&ha, ctx); /* second: done_posted guard → no-op */ + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_ctrlhello_call_count, + "done_posted guard must prevent double ctrlhello push"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_push_config_call_count, + "done_posted guard must prevent double config push"); + + free_ctx(ctx); + ha_fini(&ha); + clear_hooks(); + printf("test_done_posted_prevents_double_push: PASS " + "(ctrlhello=%d, push_config=%d)\n", + g_ctrlhello_call_count, g_push_config_call_count); +} + +/* ── Suite entry point ───────────────────────────────────────────────────── */ + +void test_ha_repl_push(void) +{ + RUN_TEST(test_committing_calls_ctrlhello_push); + RUN_TEST(test_committing_calls_push_config); + RUN_TEST(test_standby_result_does_not_push); + RUN_TEST(test_3node_win_calls_both_pushes); + RUN_TEST(test_done_posted_prevents_double_push); +}