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

feat(election-iomgr-refactor): config/WAL replication push to standbys in COMMITTING path — task 4.2

Browse source
This commit is contained in:
Rongsong Shen 2026-05-30 11:56:31 +00:00
parent f4a6726f1b
commit 757558bb33
6 changed files with 433 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

54
src/controller/cluster.c
View file

@ -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) */

41
src/controller/ha.c
View file

@ -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));

24
src/controller/ha.h
View file

@ -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

1
src/controller/meson.build
View file

@ -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',

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

@ -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();
}

311
src/controller/testsuite/test_ha_repl_push.c Normal file
View file

@ -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 <errno.h>
#include <semaphore.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unity/unity.h>
#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);
}