diff --git a/src/controller/ha.c b/src/controller/ha.c index 788597e0..894e1eb1 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -1,21 +1,29 @@ #define _GNU_SOURCE #include "ha.h" +#include #include #include +#include +#include #include #include #include #include #include #include +#include #include #include +#include +#include + /* Circular-include safety: cluster_def.h includes ha.h (include-guarded), * and ha.h only forward-declares cluster_t — no cycle. */ #include "xce/cluster_def.h" #include "persist.h" +#include "utils.h" /* ── PRNG seeding ───────────────────────────────────────────── */ @@ -69,6 +77,11 @@ int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority) atomic_store_explicit(&ha->watchdog_stop, 0, memory_order_relaxed); pthread_mutex_init(&ha->lock, NULL); + + ha->ssl_ctx = NULL; + ha->jitter_seed = 0; + ha->standby_endpoints = NULL; + return 0; } @@ -114,6 +127,231 @@ void ha_do_demote(ha_state_t *ha) /* Hook disable requires hooksys_t — done by ha_demote_to_standby(cluster_t*) */ } +/* ── P1-6: Election probing ─────────────────────────────────── */ + +/* + * probe_thread: attempt a TCP (optionally SSL) connection to one candidate. + * Sets ctx->got_ack = 1 if the connection succeeds before election_done is set. + */ +static void *probe_thread(void *arg) +{ + probe_ctx_t *ctx = (probe_ctx_t *)arg; + + struct addrinfo hints, *res = NULL; + memset(&hints, 0, sizeof(hints)); + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + + char port_str[16]; + snprintf(port_str, sizeof(port_str), "%u", (unsigned)ctx->candidate->port); + if (getaddrinfo(ctx->candidate->hostname, port_str, &hints, &res) != 0) + goto done; + + int sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol); + if (sock < 0) { freeaddrinfo(res); goto done; } + + /* Non-blocking connect */ + int flags = fcntl(sock, F_GETFL, 0); + fcntl(sock, F_SETFL, flags | O_NONBLOCK); + connect(sock, res->ai_addr, res->ai_addrlen); + freeaddrinfo(res); + res = NULL; + + /* Poll loop: check election_done between iterations */ + while (!atomic_load(ctx->election_done)) { + struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP }; + int pr = poll(&pfd, 1, (int)ctx->timeout_ms); + if (pr < 0 && errno == EINTR) continue; + if (pr == 0) goto cleanup; /* poll timeout */ + if (pfd.revents & (POLLERR | POLLHUP)) goto cleanup; + if (pfd.revents & POLLOUT) break; + } + if (atomic_load(ctx->election_done)) goto cleanup; + + /* Verify connect succeeded */ + { + int err = 0; + socklen_t el = sizeof(err); + getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el); + if (err != 0) goto cleanup; + } + + /* SSL handshake — only if ssl_ctx is set */ + if (ctx->ssl_ctx) { + SSL *ssl = SSL_new((SSL_CTX *)ctx->ssl_ctx); + if (!ssl) goto cleanup; + /* Restore blocking for SSL */ + fcntl(sock, F_SETFL, flags & ~O_NONBLOCK); + SSL_set_fd(ssl, sock); + int sr; + do { + sr = SSL_connect(ssl); + } while (sr <= 0 && + (SSL_get_error(ssl, sr) == SSL_ERROR_WANT_READ || + SSL_get_error(ssl, sr) == SSL_ERROR_WANT_WRITE) && + !atomic_load(ctx->election_done)); + SSL_shutdown(ssl); + SSL_free(ssl); + if (sr <= 0 || atomic_load(ctx->election_done)) + goto cleanup; + } + + ctx->got_ack = 1; + +cleanup: + close(sock); +done: + return NULL; +} + +/* + * wait_any_probe: spin-poll until any probe sets got_ack, or deadline passes. + * Returns 1 if any probe acknowledged, 0 if deadline expired. + */ +static int wait_any_probe(probe_ctx_t *probes, int n, + const struct timespec *deadline) +{ + while (1) { + for (int i = 0; i < n; i++) { + if (probes[i].got_ack) + return 1; + } + struct timespec now; + clock_gettime(CLOCK_MONOTONIC, &now); + if (now.tv_sec > deadline->tv_sec || + (now.tv_sec == deadline->tv_sec && + now.tv_nsec >= deadline->tv_nsec)) + return 0; + /* Sleep 10ms between checks */ + struct timespec sleep_ts = { .tv_sec = 0, .tv_nsec = 10 * 1000000L }; + nanosleep(&sleep_ts, NULL); + } +} + +/* + * am_i_primary: election probe logic. + * + * Priority-0 is unconditionally primary — never probes. + * Priority-N probes the N candidates with lower index (higher priority). + * If any responds before election_timeout + priority_step*N + jitter, this + * node becomes standby. If deadline expires with no response, self-promote. + */ +int am_i_primary(cluster_t *c) +{ + /* Priority-0 never probes — it is unconditionally primary */ + if (c->ha.my_priority == 0) + return 1; + + const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg; + int n_probes = (c->ha.my_priority < pcfg->candidates_count) + ? c->ha.my_priority + : pcfg->candidates_count; + if (n_probes == 0) + return 1; + + /* Seed jitter PRNG if needed */ + if (c->ha.jitter_seed == 0) + ha_seed_jitter(&c->ha.jitter_seed, c->ha.my_priority); + + atomic_int election_done; + atomic_init(&election_done, 0); + probe_ctx_t probes[MAX_HA_CANDIDATES]; + + for (int i = 0; i < n_probes; i++) { + probes[i].candidate = (xce_candidate_t *)&pcfg->candidates[i]; + probes[i].ssl_ctx = c->ha.ssl_ctx; + probes[i].timeout_ms = c->ha.config.probe_connect_timeout_ms; + probes[i].election_done = &election_done; + probes[i].got_ack = 0; + pthread_create(&probes[i].thread, NULL, probe_thread, &probes[i]); + } + + /* Compute deadline = election_timeout + priority_step * my_priority + jitter */ + long to_ms = (long)c->ha.config.election_timeout_ms + + (long)c->ha.my_priority + * (long)c->ha.config.election_priority_step_ms; + if (c->ha.config.election_jitter_max_ms > 0) { + to_ms += (long)(xorshift64_next(&c->ha.jitter_seed) + % ((uint64_t)c->ha.config.election_jitter_max_ms + 1)); + } + + struct timespec deadline; + clock_gettime(CLOCK_MONOTONIC, &deadline); + deadline.tv_sec += to_ms / 1000; + deadline.tv_nsec += (to_ms % 1000) * 1000000L; + if (deadline.tv_nsec >= 1000000000L) { + deadline.tv_sec++; + deadline.tv_nsec -= 1000000000L; + } + + int got_ack = wait_any_probe(probes, n_probes, &deadline); + + /* Signal all threads to stop */ + atomic_store(&election_done, 1); + for (int i = 0; i < n_probes; i++) + pthread_join(probes[i].thread, NULL); + + if (got_ack) { + WARN("HA election: higher-priority candidate responded — " + "starting as standby (my_priority=%d)", c->ha.my_priority); + return 0; + } + INFO("HA election: no higher-priority candidate responded within %ldms — " + "starting as primary (my_priority=%d)", to_ms, c->ha.my_priority); + return 1; +} + +/* + * ha_is_primary: lock-free role check. + */ +int ha_is_primary(const ha_state_t *ha) +{ + return (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire) + == CTRL_PRIMARY; +} + +/* + * ha_check_priority_stepdown: called from heartbeat handler on simultaneous boot + * (both nodes at term=0). Steps down PRIMARY to STANDBY if a higher-priority peer + * (lower index = lower sender_priority value) is observed. + */ +void ha_check_priority_stepdown(ha_state_t *ha, int sender_priority, + uint64_t sender_term) +{ + /* Only act if sender has strictly higher priority (lower index) */ + if (sender_priority >= ha->my_priority) + return; + + uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire); + if (my_term > 0 || sender_term > 0) + return; /* term-based election took over — don't interfere */ + + ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role, + memory_order_acquire); + if (role == CTRL_PRIMARY) { + WARN("HA: stepping down — higher-priority peer (priority=%d) present at term=0", + sender_priority); + ha_do_demote(ha); + } +} + +/* + * start_heartbeat_listener / stop_heartbeat_listener: + * Stub implementations — full listener is implemented in Task 7 (cluster.c). + * The stubs allow am_i_primary() callers (main.c / cluster.c) to compile and + * link without requiring the full Phase-2 network stack. + */ +void start_heartbeat_listener(cluster_t *c) +{ + (void)c; + /* Heartbeat listener stub — full implementation in cluster.c Task 7 */ +} + +void stop_heartbeat_listener(cluster_t *c) +{ + (void)c; +} + /* ── P3-6: Test-only failover shim ────────────────────────────── */ #ifdef XCE_TESTING @@ -238,12 +476,6 @@ int ha_no_primary_known(const ha_state_t *ha) return (ha->primary_host[0] == '\0' || ha->primary_port == 0) ? 1 : 0; } -/* priority-0 node is primary; full election probing is a future enhancement */ -int am_i_primary(cluster_t *c) -{ - return c->ha.my_priority == 0; -} - /* * ha_promote_to_primary: increment term, set role=PRIMARY, then register * persistence hooks (idempotency guard via hooks_registered). diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 3193c7e0..eb0af1f8 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_am_i_primary(void); extern void test_ha_cluster(void); extern void test_ha_redirect(void); extern void test_p1_integration(void); @@ -121,6 +122,7 @@ int main(int argc, char* argv[]) TEST(test_e2e_gc); TEST(test_e2e_capnp); TEST(test_ha_role); + TEST(test_am_i_primary); TEST(test_ha_cluster); TEST(test_ha_redirect); TEST(test_p1_integration); diff --git a/src/controller/testsuite/test_ha_role.c b/src/controller/testsuite/test_ha_role.c index 7a05d6e5..33e2036d 100644 --- a/src/controller/testsuite/test_ha_role.c +++ b/src/controller/testsuite/test_ha_role.c @@ -7,6 +7,8 @@ #include #include "ha.h" +#include "myconf.h" +#include "xce/cluster_def.h" static void test_ha_init_default_role_is_standby(void) { @@ -232,6 +234,92 @@ static void test_sigusr1_sets_failover_requested(void) printf("test_sigusr1_sets_failover_requested: PASS\n"); } +/* ── P1-6: am_i_primary() election tests ────────────────────────────── */ + +/* test: priority-0 with no peers → immediately primary (no probing needed) */ +void test_am_i_primary_p0_no_peers(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + myconf_t myconf; + memset(&myconf, 0, sizeof(myconf)); + c.myconf = &myconf; + + ha_config_t cfg = { + .heartbeat_interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS, + .heartbeat_timeout_ms = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS, + .election_timeout_ms = 500, /* minimum valid value */ + .probe_connect_timeout_ms = 100, + .max_promotion_retries = 1, + .election_jitter_max_ms = 0, + .election_priority_step_ms = 0, + }; + TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &cfg, 0)); + TEST_ASSERT_EQUAL_INT(1, am_i_primary(&c)); + ha_fini(&c.ha); + printf("test_am_i_primary_p0_no_peers: PASS\n"); +} + +/* test: priority-0 with peers → still primary (never probes higher-priority nodes) */ +void test_am_i_primary_p0_with_peers(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + myconf_t myconf; + memset(&myconf, 0, sizeof(myconf)); + myconf.peers_cfg.candidates_count = 2; + strncpy(myconf.peers_cfg.candidates[0].hostname, "127.0.0.1", XCE_MAXNAME_LEN - 1); + myconf.peers_cfg.candidates[0].port = 19900; + strncpy(myconf.peers_cfg.candidates[1].hostname, "127.0.0.1", XCE_MAXNAME_LEN - 1); + myconf.peers_cfg.candidates[1].port = 19901; + c.myconf = &myconf; + + ha_config_t cfg = { + .heartbeat_interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS, + .heartbeat_timeout_ms = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS, + .election_timeout_ms = 500, /* minimum valid value */ + .probe_connect_timeout_ms = 100, + .max_promotion_retries = 1, + .election_jitter_max_ms = 0, + .election_priority_step_ms = 0, + }; + TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &cfg, 0)); + TEST_ASSERT_EQUAL_INT(1, am_i_primary(&c)); + ha_fini(&c.ha); + printf("test_am_i_primary_p0_with_peers: PASS\n"); +} + +/* test: priority-1 with no higher-priority peer alive → deadline expires → primary */ +void test_am_i_primary_p1_deadline_expires(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + myconf_t myconf; + memset(&myconf, 0, sizeof(myconf)); + /* Point to ports where nothing is listening */ + myconf.peers_cfg.candidates_count = 2; + strncpy(myconf.peers_cfg.candidates[0].hostname, "127.0.0.1", XCE_MAXNAME_LEN - 1); + myconf.peers_cfg.candidates[0].port = 19910; /* nothing listening */ + strncpy(myconf.peers_cfg.candidates[1].hostname, "127.0.0.1", XCE_MAXNAME_LEN - 1); + myconf.peers_cfg.candidates[1].port = 19911; + c.myconf = &myconf; + + ha_config_t cfg = { + .heartbeat_interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS, + .heartbeat_timeout_ms = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS, + .election_timeout_ms = 600, /* expires quickly; > 500ms floor */ + .probe_connect_timeout_ms = 200, + .max_promotion_retries = 1, + .election_jitter_max_ms = 0, + .election_priority_step_ms = 0, + }; + TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &cfg, 1)); /* priority 1 — must probe priority 0 */ + int result = am_i_primary(&c); + TEST_ASSERT_EQUAL_INT(1, result); /* no peer responded → self-declare primary */ + ha_fini(&c.ha); + printf("test_am_i_primary_p1_deadline_expires: PASS\n"); +} + void test_ha_role(void) { RUN_TEST(test_ha_init_default_role_is_standby); @@ -247,3 +335,12 @@ void test_ha_role(void) RUN_TEST(test_demote_sets_standby_role); RUN_TEST(test_sigusr1_sets_failover_requested); } + +/* ── P1-6 suite: am_i_primary() election ─────────────────────────────── */ + +void test_am_i_primary(void) +{ + RUN_TEST(test_am_i_primary_p0_no_peers); + RUN_TEST(test_am_i_primary_p0_with_peers); + RUN_TEST(test_am_i_primary_p1_deadline_expires); +}