From c627f69d503abcc9e3267fca6ceffd5c22b4fed5 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 7 May 2026 14:55:48 +0000 Subject: [PATCH] feat: wire ha_state_t into cluster_t + HA startup/shutdown lifecycle (P1-5) --- src/controller/cluster.c | 36 +++++ src/controller/ha.c | 63 +++++++-- src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + src/controller/testsuite/test_ha_cluster.c | 148 +++++++++++++++++++++ src/include/xce/cluster_def.h | 4 + 6 files changed, 241 insertions(+), 13 deletions(-) create mode 100644 src/controller/testsuite/test_ha_cluster.c diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 2bc24c8c..f5c62185 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -632,6 +632,32 @@ int cluster_init(cluster_t* c) } } + /* HA initialisation — P1-5 */ + { + ha_config_t ha_cfg = { + .heartbeat_interval_ms = 3000, + .heartbeat_timeout_ms = 10000, + .election_timeout_ms = 2000, + .probe_connect_timeout_ms = 500, + .max_promotion_retries = 5, + }; + /* my_priority: single-node P1 always uses 0 (highest priority) */ + rc = ha_init(&c->ha, &ha_cfg, 0); + if (rc != 0) { + ERROR("ha_init failed: %d", rc); + persist_fini(&c->persist); + return rc; + } + ha_install_sigusr1(&c->ha); + + /* Determine initial role: priority-0 node starts as primary */ + if (am_i_primary(c)) + ha_promote_to_primary(c); + else + atomic_store_explicit(&c->ha.role, (int32_t)CTRL_STANDBY, + memory_order_release); + } + xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT, HOOK_DETAIL_CLUSTER_INIT, c); @@ -787,6 +813,13 @@ int cluster_start(cluster_t* c) return rc; } + /* Start HA heartbeat watchdog thread — P1-5 */ + rc = ha_start_watchdog(&c->ha); + if (rc != 0) { + ERROR("ha_start_watchdog failed: %d", rc); + return rc; + } + delay_task(&(c->delaymgr), c->myconf->cluster_config->heartbeat_interval, -1, (delay_task_entry_fn_t)check_workers_timeout, (void*)c); @@ -819,6 +852,9 @@ int cluster_run(cluster_t* c) void cluster_fini(cluster_t* c) { + /* Stop HA watchdog thread before tearing down other subsystems — P1-5 */ + ha_fini(&c->ha); + /* Stop persistence before tearing down the rest */ persist_stop(&c->persist); persist_fini(&c->persist); diff --git a/src/controller/ha.c b/src/controller/ha.c index 879a407e..3ce90e52 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -9,6 +9,17 @@ #include #include +/* + * P1-5: include full cluster_t definition (with embedded ha_state_t ha) + * and persist hook function declarations so ha_promote_to_primary / + * ha_demote_to_standby can dereference c->ha and manage persist hooks. + * + * Circular-include safety: cluster_def.h includes ha.h (which has include + * guards), and ha.h only forward-declares cluster_t — no cycle. + */ +#include "xce/cluster_def.h" +#include "persist.h" + int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority) { if (!ha || !cfg) @@ -176,31 +187,57 @@ void ha_fini(ha_state_t *ha) } /* - * The three cluster_t-dependent stubs below cannot dereference c->ha until - * P1-5 adds `ha_state_t ha` to cluster_t. They compile as no-ops now and - * will be fleshed out in P1-3a after P1-5 wires the field in. + * P1-5: cluster_t now embeds ha_state_t ha, so the three functions below + * can dereference c->ha. */ +/* + * am_i_primary: priority-0 node is primary for single-node P1. + * Full election probing added in P1-3 (future). + */ int am_i_primary(cluster_t *c) { - (void)c; - /* P1-3 adds real election probing. - * Stub: always report not-primary until cluster wiring is in place. */ - return 0; + return c->ha.my_priority == 0; } +/* + * ha_promote_to_primary: increment term, set role=PRIMARY, then register + * persistence hooks (idempotency guard via hooks_registered). + * Safe to call on a zero-init cluster_t (zeroed hooksys → xce_add_hook + * returns -EINVAL harmlessly; hooks_registered is still set to 1). + */ int ha_promote_to_primary(cluster_t *c) { - (void)c; - /* Full sequence in P1-3a (WAL flush, hook calls, peer notify). - * Stub: no-op — role management moved to P1-3a. */ + ha_do_promote(&c->ha); + + /* Register persistence hooks — idempotency guard prevents double-add */ + if (c->ha.hooks_registered == 0) { + xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_JOB, + (hook_fn_t)persist_job_hook_fun, &c->persist); + xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_QUEUE, + (hook_fn_t)persist_queue_hook_fun, &c->persist); + xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER, + (hook_fn_t)persist_worker_hook_fun, &c->persist); + c->ha.hooks_registered = 1; + } return 0; } +/* + * ha_demote_to_standby: disable persistence hooks, then set role=STANDBY + * and reset the hooks_registered idempotency guard (via ha_do_demote). + */ int ha_demote_to_standby(cluster_t *c) { - (void)c; - /* Full sequence in P1-3a. - * Stub: no-op — role management moved to P1-3a. */ + /* Disable persistence hooks before changing role */ + if (c->ha.hooks_registered) { + xce_disable_hook(&c->hooksys, XCE_CONTROLLER_HOOK_JOB, + (hook_fn_t)persist_job_hook_fun); + xce_disable_hook(&c->hooksys, XCE_CONTROLLER_HOOK_QUEUE, + (hook_fn_t)persist_queue_hook_fun); + xce_disable_hook(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER, + (hook_fn_t)persist_worker_hook_fun); + } + ha_do_demote(&c->ha); /* sets role=STANDBY, resets hooks_registered=0 */ return 0; } diff --git a/src/controller/meson.build b/src/controller/meson.build index f8b52123..4bf76517 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -84,6 +84,7 @@ controller_unit_test_src = [ 'testsuite/test_e2e_gc.c', 'testsuite/test_e2e_capnp.c', 'testsuite/test_ha_role.c', + 'testsuite/test_ha_cluster.c', 'persist.c', 'ha.c', ] diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 1904b622..c72e8e21 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -20,6 +20,7 @@ extern void test_e2e_hooks(void); extern void test_e2e_gc(void); extern void test_e2e_capnp(void); extern void test_ha_role(void); +extern void test_ha_cluster(void); void setUp(void) { return; } @@ -109,6 +110,7 @@ int main(int argc, char* argv[]) TEST(test_e2e_gc); TEST(test_e2e_capnp); TEST(test_ha_role); + TEST(test_ha_cluster); return UNITY_END(); } diff --git a/src/controller/testsuite/test_ha_cluster.c b/src/controller/testsuite/test_ha_cluster.c new file mode 100644 index 00000000..d64d810a --- /dev/null +++ b/src/controller/testsuite/test_ha_cluster.c @@ -0,0 +1,148 @@ +/* + * test_ha_cluster.c — P1-5: structural tests for ha_state_t embedded in cluster_t + * + * These tests validate that: + * 1. cluster_t has an ha_state_t ha field (compile-time structural check). + * 2. ha_promote_to_primary() and ha_demote_to_standby() use c->ha (not no-ops). + * + * Tests do NOT call cluster_init() — that requires a live config/SSL/network stack. + * Instead they manually zero-init cluster_t and call ha_init() on c.ha directly. + */ + +#include +#include + +#include + +#include "ha.h" +#include "xce/cluster_def.h" + +/* Valid HA config shared across 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, +}; + +/* + * test_cluster_t_has_ha_field: + * Compile-time structural check — cluster_t must embed ha_state_t as field `ha`. + * If `c.ha` does not exist, this file will not compile → RED. + */ +static void test_cluster_t_has_ha_field(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + + /* Access the field — compile error if absent */ + (void)atomic_load_explicit(&c.ha.role, memory_order_acquire); + + printf("test_cluster_t_has_ha_field: PASS\n"); +} + +/* + * test_ha_promote_to_primary_uses_c_ha: + * After ha_init(&c.ha, ...) and calling ha_promote_to_primary(&c), the role + * stored in c.ha must be CTRL_PRIMARY (not still CTRL_STANDBY). + * This would fail if ha_promote_to_primary() is still the (void)c stub. + */ +static void test_ha_promote_to_primary_uses_c_ha(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + + /* Manually initialise only the HA field — no cluster_init() required */ + TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &g_cfg, 0 /* my_priority = 0 → primary */)); + TEST_ASSERT_EQUAL_INT(CTRL_STANDBY, + (int)atomic_load_explicit(&c.ha.role, memory_order_acquire)); + + /* c.hooksys is zeroed — hooks_registered guard prevents xce_add_hook() call */ + ha_promote_to_primary(&c); + + TEST_ASSERT_EQUAL_INT(CTRL_PRIMARY, + (int)atomic_load_explicit(&c.ha.role, memory_order_acquire)); + + ha_fini(&c.ha); + printf("test_ha_promote_to_primary_uses_c_ha: PASS\n"); +} + +/* + * test_ha_demote_to_standby_uses_c_ha: + * Promote then demote. Role must end as CTRL_STANDBY. + */ +static void test_ha_demote_to_standby_uses_c_ha(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + + TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &g_cfg, 0)); + + ha_promote_to_primary(&c); + TEST_ASSERT_EQUAL_INT(CTRL_PRIMARY, + (int)atomic_load_explicit(&c.ha.role, memory_order_acquire)); + + ha_demote_to_standby(&c); + TEST_ASSERT_EQUAL_INT(CTRL_STANDBY, + (int)atomic_load_explicit(&c.ha.role, memory_order_acquire)); + + ha_fini(&c.ha); + printf("test_ha_demote_to_standby_uses_c_ha: PASS\n"); +} + +/* + * test_ha_promote_increments_term_via_cluster: + * Each ha_promote_to_primary() call must increment the election term stored + * in c.ha.term, because ha_do_promote() is called inside. + */ +static void test_ha_promote_increments_term_via_cluster(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + + TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &g_cfg, 0)); + uint64_t term0 = atomic_load_explicit(&c.ha.term, memory_order_acquire); + + ha_promote_to_primary(&c); + uint64_t term1 = atomic_load_explicit(&c.ha.term, memory_order_acquire); + TEST_ASSERT_TRUE(term1 > term0); + + ha_fini(&c.ha); + printf("test_ha_promote_increments_term_via_cluster: PASS\n"); +} + +/* + * test_hooks_registered_guard_prevents_double_register: + * When hooksys is zeroed (no real hook system) and hooks_registered is 0, + * ha_promote_to_primary() must set hooks_registered = 1 so a second + * call skips the registration path. + */ +static void test_hooks_registered_guard_prevents_double_register(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + + TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &g_cfg, 0)); + TEST_ASSERT_EQUAL_INT(0, c.ha.hooks_registered); + + ha_promote_to_primary(&c); + /* hooks_registered must now be 1 (idempotency guard set) */ + TEST_ASSERT_EQUAL_INT(1, c.ha.hooks_registered); + + /* Second promote: must not crash even with zeroed hooksys */ + ha_promote_to_primary(&c); + TEST_ASSERT_EQUAL_INT(1, c.ha.hooks_registered); + + ha_fini(&c.ha); + printf("test_hooks_registered_guard_prevents_double_register: PASS\n"); +} + +void test_ha_cluster(void) +{ + RUN_TEST(test_cluster_t_has_ha_field); + RUN_TEST(test_ha_promote_to_primary_uses_c_ha); + RUN_TEST(test_ha_demote_to_standby_uses_c_ha); + RUN_TEST(test_ha_promote_increments_term_via_cluster); + RUN_TEST(test_hooks_registered_guard_prevents_double_register); +} diff --git a/src/include/xce/cluster_def.h b/src/include/xce/cluster_def.h index dc27d63e..ed7ae3d8 100644 --- a/src/include/xce/cluster_def.h +++ b/src/include/xce/cluster_def.h @@ -16,6 +16,9 @@ #include "netio.h" #include "persist.h" #include "vtaskmgr.h" + +/* ha.h forward-declares cluster_t; including it here is safe (no circular dep) */ +#include "ha.h" #include "xce/types.h" #include "xlist.h" #include "zarray.h" @@ -116,6 +119,7 @@ typedef struct cluster_s { cbuf_t wbuf; int running; persist_t persist; /* Persistence module — P0-F */ + ha_state_t ha; /* HA role management — P1-5 */ } cluster_t; #endif /* XCE_CLUSTER_DEF_H */