forked from lei.chen/myxce
1688 lines
61 KiB
C
1688 lines
61 KiB
C
#define _GNU_SOURCE
|
||
#include "ha.h"
|
||
|
||
#include <arpa/inet.h>
|
||
#include <errno.h>
|
||
#include <fcntl.h>
|
||
#include <inttypes.h>
|
||
#include <netdb.h>
|
||
#include <poll.h>
|
||
#include <pthread.h>
|
||
#include <signal.h>
|
||
#include <stdatomic.h>
|
||
#include <stdio.h>
|
||
#include <stdlib.h>
|
||
#include <string.h>
|
||
#include <sys/epoll.h>
|
||
#include <sys/socket.h>
|
||
#include <time.h>
|
||
#include <unistd.h>
|
||
#include <zlib.h>
|
||
|
||
#include <openssl/ssl.h>
|
||
#include <openssl/err.h>
|
||
|
||
/* 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 "xce/errnox.h"
|
||
#include "controller.capnp.h"
|
||
#include "persist.h"
|
||
#include "utils.h"
|
||
|
||
/* ── HA dispatch gate error codes (defined here; declared extern in errnox.h) ── */
|
||
int XCE_ERR_NOT_READY = E_XCE_NOT_READY;
|
||
int XCE_ERR_STANDBY = E_XCE_STANDBY;
|
||
|
||
/* ── PRNG seeding ───────────────────────────────────────────── */
|
||
|
||
void ha_seed_jitter(uint64_t *seed, int my_priority)
|
||
{
|
||
uint8_t buf[8] = {0};
|
||
int fd = open("/dev/urandom", O_RDONLY);
|
||
if (fd >= 0) {
|
||
ssize_t n = 0;
|
||
while (n < 8) {
|
||
ssize_t r = read(fd, buf + n, (size_t)(8 - n));
|
||
if (r <= 0) break;
|
||
n += r;
|
||
}
|
||
close(fd);
|
||
if (n == 8) {
|
||
memcpy(seed, buf, 8);
|
||
if (*seed != 0) return;
|
||
}
|
||
}
|
||
struct timespec ts;
|
||
clock_gettime(CLOCK_MONOTONIC, &ts);
|
||
*seed = (uint64_t)ts.tv_nsec
|
||
^ ((uint64_t)my_priority << 56)
|
||
^ ((uint64_t)getpid() << 16);
|
||
}
|
||
|
||
int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority)
|
||
{
|
||
if (!ha || !cfg)
|
||
return -EINVAL;
|
||
|
||
/* Zero first so ha_fini() is safe even if validation fails below */
|
||
memset(ha, 0, sizeof(*ha));
|
||
|
||
if (cfg->heartbeat_interval_ms < 2000)
|
||
return -EINVAL;
|
||
if (cfg->heartbeat_timeout_ms < cfg->heartbeat_interval_ms)
|
||
return -EINVAL;
|
||
if (cfg->election_timeout_ms < 500)
|
||
return -EINVAL;
|
||
ha->config = *cfg;
|
||
ha->my_priority = my_priority;
|
||
|
||
atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY, memory_order_release);
|
||
atomic_store_explicit(&ha->term, 0ULL, memory_order_release);
|
||
atomic_store_explicit(&ha->applied_lsn, 0ULL, memory_order_release);
|
||
atomic_store_explicit(&ha->promotion_failed, 0, memory_order_relaxed);
|
||
atomic_store_explicit(&ha->promotion_retry_count, 0, memory_order_relaxed);
|
||
atomic_store_explicit(&ha->in_catch_up, 0, memory_order_relaxed);
|
||
atomic_store_explicit(&ha->failover_requested, 0, memory_order_relaxed);
|
||
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;
|
||
ha->primary_ctrl_fd = -1;
|
||
|
||
/* heartbeat_seen starts at 0: no heartbeat seen yet.
|
||
* ha_wait_for_heartbeat_or_timeout() calls sem_timedwait on this. */
|
||
sem_init(&ha->heartbeat_seen, 0, 0);
|
||
|
||
return 0;
|
||
}
|
||
|
||
/* ── Promotion / demotion / SIGUSR1 ────────────────────────── */
|
||
|
||
/* Async-signal-safe: atomic store on int-width type (POSIX.1-2017 §2.4.3).
|
||
* Pointer set once before signals are enabled — no locking required. */
|
||
static ha_state_t *g_sigusr1_ha = NULL;
|
||
|
||
static void sigusr1_handler(int sig)
|
||
{
|
||
(void)sig;
|
||
if (g_sigusr1_ha)
|
||
atomic_store_explicit(&g_sigusr1_ha->failover_requested, 1,
|
||
memory_order_release);
|
||
}
|
||
|
||
void ha_install_sigusr1(ha_state_t *ha)
|
||
{
|
||
g_sigusr1_ha = ha;
|
||
struct sigaction sa;
|
||
memset(&sa, 0, sizeof(sa));
|
||
sa.sa_handler = sigusr1_handler;
|
||
sigemptyset(&sa.sa_mask);
|
||
sa.sa_flags = SA_RESTART;
|
||
sigaction(SIGUSR1, &sa, NULL);
|
||
}
|
||
|
||
void ha_do_promote(ha_state_t *ha)
|
||
{
|
||
/* Monotonically increasing fencing token — standbys reject entries from lower terms */
|
||
atomic_fetch_add_explicit(&ha->term, 1, memory_order_acq_rel);
|
||
atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY, memory_order_release);
|
||
/* COH-5: new primary starts a fresh CRC chain; standbys reset their chain on catch-up */
|
||
ha->last_received_crc = WAL_CRC_NONE;
|
||
/* Req 10.5: log FSM transition to PRIMARY */
|
||
INFO("HA FSM: → CTRL_PRIMARY (term=%" PRIu64 ")",
|
||
atomic_load_explicit(&ha->term, memory_order_acquire));
|
||
}
|
||
|
||
void ha_do_demote(ha_state_t *ha)
|
||
{
|
||
atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY, memory_order_release);
|
||
/* Reset so the next ha_promote_to_primary() re-registers hooks idempotently */
|
||
ha->hooks_registered = 0;
|
||
/* Hook disable requires hooksys_t — done by ha_demote_to_standby(cluster_t*) */
|
||
/* Req 10.5: log FSM transition to STANDBY */
|
||
INFO("HA FSM: → CTRL_STANDBY (term=%" PRIu64 ")",
|
||
atomic_load_explicit(&ha->term, memory_order_acquire));
|
||
}
|
||
|
||
/* ── 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);
|
||
if (sr <= 0) {
|
||
int ssl_err = SSL_get_error(ssl, sr);
|
||
if (ssl_err != SSL_ERROR_WANT_READ &&
|
||
ssl_err != SSL_ERROR_WANT_WRITE)
|
||
break;
|
||
}
|
||
} while (sr <= 0 && !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() deleted — replaced by quorum election (Req 2.1) */
|
||
|
||
/*
|
||
* 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
|
||
uint64_t ha_trigger_manual_failover_for_test(ha_state_t *ha)
|
||
{
|
||
ha_do_demote(ha);
|
||
ha_do_promote(ha);
|
||
return atomic_load_explicit(&ha->term, memory_order_acquire);
|
||
}
|
||
#endif
|
||
|
||
/* ha_check_heartbeat_nocluster() removed from the public API — logic inlined in
|
||
* ha_watchdog_thread(). XCE_TESTING shim preserved so existing unit tests compile. */
|
||
#ifdef XCE_TESTING
|
||
void ha_check_heartbeat_nocluster(ha_state_t *ha)
|
||
{
|
||
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
|
||
if (role != CTRL_STANDBY) return;
|
||
if (ha->primary_host[0] == '\0') return;
|
||
|
||
struct timespec now;
|
||
clock_gettime(CLOCK_MONOTONIC, &now);
|
||
pthread_mutex_lock(&ha->lock);
|
||
struct timespec last = ha->last_heartbeat;
|
||
pthread_mutex_unlock(&ha->lock);
|
||
|
||
uint64_t elapsed_ms =
|
||
(uint64_t)(now.tv_sec - last.tv_sec) * 1000ULL +
|
||
(uint64_t)(now.tv_nsec - last.tv_nsec) / 1000000ULL;
|
||
|
||
if (elapsed_ms >= ha->config.heartbeat_timeout_ms)
|
||
atomic_store_explicit(&ha->failover_requested, 1, memory_order_release);
|
||
}
|
||
#endif /* XCE_TESTING */
|
||
|
||
static void *ha_watchdog_thread(void *arg)
|
||
{
|
||
ha_state_t *ha = arg;
|
||
|
||
while (!atomic_load_explicit(&ha->watchdog_stop, memory_order_acquire)) {
|
||
long ms = (long)ha->config.heartbeat_interval_ms;
|
||
struct timespec ts = {
|
||
.tv_sec = ms / 1000,
|
||
.tv_nsec = (ms % 1000) * 1000000L,
|
||
};
|
||
nanosleep(&ts, NULL);
|
||
|
||
/* Re-check stop flag after sleep */
|
||
if (atomic_load_explicit(&ha->watchdog_stop, memory_order_acquire)) break;
|
||
|
||
/* Heartbeat timeout check (inlined from deleted ha_check_heartbeat_nocluster) */
|
||
{
|
||
ctrl_role_t role =
|
||
(ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
|
||
if (role == CTRL_STANDBY && ha->primary_host[0] != '\0') {
|
||
struct timespec now;
|
||
clock_gettime(CLOCK_MONOTONIC, &now);
|
||
pthread_mutex_lock(&ha->lock);
|
||
struct timespec last = ha->last_heartbeat;
|
||
pthread_mutex_unlock(&ha->lock);
|
||
uint64_t elapsed_ms =
|
||
(uint64_t)(now.tv_sec - last.tv_sec) * 1000ULL +
|
||
(uint64_t)(now.tv_nsec - last.tv_nsec) / 1000000ULL;
|
||
if (elapsed_ms >= ha->config.heartbeat_timeout_ms)
|
||
atomic_store_explicit(&ha->failover_requested, 1,
|
||
memory_order_release);
|
||
}
|
||
}
|
||
|
||
if (atomic_load_explicit(&ha->failover_requested, memory_order_acquire)) {
|
||
atomic_store_explicit(&ha->failover_requested, 0, memory_order_release);
|
||
/* Promotion requires cluster_t; the watchdog only has ha_state_t.
|
||
* Full ha_promote_to_primary(c) is called from the cluster's own
|
||
* failover path which holds cluster_t context. */
|
||
}
|
||
}
|
||
return NULL;
|
||
}
|
||
|
||
int ha_start_watchdog(ha_state_t *ha)
|
||
{
|
||
atomic_store_explicit(&ha->watchdog_stop, 0, memory_order_release);
|
||
int rc = pthread_create(&ha->watchdog_thread, NULL, ha_watchdog_thread, ha);
|
||
if (rc != 0) return -rc;
|
||
return 0;
|
||
}
|
||
|
||
void ha_stop_watchdog(ha_state_t *ha)
|
||
{
|
||
atomic_store_explicit(&ha->watchdog_stop, 1, memory_order_release);
|
||
if (ha->watchdog_thread != 0) {
|
||
pthread_join(ha->watchdog_thread, NULL);
|
||
ha->watchdog_thread = 0;
|
||
}
|
||
}
|
||
|
||
void ha_fini(ha_state_t *ha)
|
||
{
|
||
if (!ha)
|
||
return;
|
||
ha_stop_watchdog(ha); /* join watchdog thread before destroying mutex */
|
||
pthread_mutex_destroy(&ha->lock);
|
||
sem_destroy(&ha->heartbeat_seen);
|
||
|
||
/* Fix ASAN leak: free the SSL context if one was allocated */
|
||
if (ha->ssl_ctx) {
|
||
SSL_CTX_free((SSL_CTX *)ha->ssl_ctx);
|
||
ha->ssl_ctx = NULL;
|
||
}
|
||
}
|
||
|
||
/* ── Dispatch gate (req 1.3, 1.4) ──────────────────────────────────────────
|
||
*
|
||
* Called from myreq_handler() in cluster.c before the opcode switch.
|
||
* Returns 0 when the operation is permitted for the current role.
|
||
*
|
||
* Primary-only ops are any op that is not one of the always-allowed set:
|
||
* nop, reply, hello, join, shutdown, heartbeat, redirect,
|
||
* steppingDown, replEntry, replAck, replSnapshot, replCatchUpDone,
|
||
* replConfig, replConfigPush, requestVote, probeConfigTag,
|
||
* walAck, upgradeConfigAck, snapshotTransfer, walCatchupEntry.
|
||
*
|
||
* For CTRL_STANDBY the check is: op is in the primary-only set → deny.
|
||
* For CTRL_INIT / CTRL_ELECTING the allowlist is tighter: only
|
||
* OP_requestVote and OP_probeConfigTag pass through.
|
||
*/
|
||
int ha_dispatch_gate(ha_state_t *ha, int opcode)
|
||
{
|
||
ctrl_role_t role =
|
||
(ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
|
||
|
||
switch (role) {
|
||
case CTRL_PRIMARY:
|
||
return 0;
|
||
|
||
case CTRL_INIT:
|
||
/* Allow election probes, config probes, AND heartbeats so the primary's
|
||
* periodic heartbeat can reach the handler and post heartbeat_seen,
|
||
* triggering the CTRL_INIT → CTRL_STANDBY transition (Req 1.1). */
|
||
if (opcode == OPCode_requestVote ||
|
||
opcode == OPCode_probeConfigTag ||
|
||
opcode == OPCode_heartbeat)
|
||
return 0;
|
||
return -XCE_ERR_NOT_READY;
|
||
|
||
case CTRL_ELECTING:
|
||
if (opcode == OPCode_requestVote || opcode == OPCode_probeConfigTag)
|
||
return 0;
|
||
return -XCE_ERR_NOT_READY;
|
||
|
||
case CTRL_STANDBY: {
|
||
/* Allow ops that standbys legitimately handle */
|
||
switch (opcode) {
|
||
case OPCode_nop:
|
||
case OPCode_reply:
|
||
case OPCode_hello:
|
||
case OPCode_join:
|
||
case OPCode_shutdown:
|
||
case OPCode_heartbeat:
|
||
case OPCode_redirect:
|
||
case OPCode_steppingDown:
|
||
case OPCode_replEntry:
|
||
case OPCode_replAck:
|
||
case OPCode_replSnapshot:
|
||
case OPCode_replCatchUpDone:
|
||
case OPCode_replConfig:
|
||
case OPCode_replConfigPush:
|
||
case OPCode_requestVote:
|
||
case OPCode_probeConfigTag:
|
||
case OPCode_walAck:
|
||
case OPCode_upgradeConfigAck:
|
||
case OPCode_snapshotTransfer:
|
||
case OPCode_walCatchupEntry:
|
||
return 0;
|
||
default:
|
||
return -XCE_ERR_STANDBY;
|
||
}
|
||
}
|
||
|
||
default:
|
||
return 0;
|
||
}
|
||
}
|
||
|
||
/* ── Redirect helpers used by join.c ───────────────────────── */
|
||
|
||
/*
|
||
* ha_should_redirect: returns 1 when this standby knows the primary's
|
||
* address and should redirect incoming join requests there.
|
||
* Returns 0 for PRIMARY role, or for STANDBY with no known primary yet.
|
||
*/
|
||
int ha_should_redirect(const ha_state_t *ha)
|
||
{
|
||
ctrl_role_t role =
|
||
(ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
|
||
if (role == CTRL_PRIMARY)
|
||
return 0;
|
||
if (ha->primary_host[0] == '\0' || ha->primary_port == 0)
|
||
return 0;
|
||
return 1;
|
||
}
|
||
|
||
/*
|
||
* ha_no_primary_known: returns 1 when this node is a STANDBY but does not
|
||
* yet know the primary's address. Callers should reply E_XCE_NOMASTER so
|
||
* the client retries rather than treating the error as permanent.
|
||
*/
|
||
int ha_no_primary_known(const ha_state_t *ha)
|
||
{
|
||
ctrl_role_t role =
|
||
(ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
|
||
if (role == CTRL_PRIMARY)
|
||
return 0;
|
||
return (ha->primary_host[0] == '\0' || ha->primary_port == 0) ? 1 : 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)
|
||
{
|
||
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)
|
||
{
|
||
/* 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;
|
||
}
|
||
|
||
/* ── HA Config Bootstrap: startup heartbeat wait (Req 1.1, 1.2) ─────────── */
|
||
|
||
/*
|
||
* ha_wait_for_heartbeat_or_timeout: called from the startup path after
|
||
* ha_init() when the node needs to discover whether a primary already exists.
|
||
*
|
||
* Blocks on sem_timedwait(&ha->heartbeat_seen, &deadline) where
|
||
* deadline = now(CLOCK_REALTIME) + election_timeout_ms + jitter
|
||
*
|
||
* The heartbeat handler (cluster.c) posts heartbeat_seen when it receives
|
||
* OPCode_heartbeat and ha->role == CTRL_INIT. ha_dispatch_gate() permits
|
||
* OPCode_heartbeat in CTRL_INIT state so the packet reaches the handler.
|
||
*
|
||
* Returns CTRL_STANDBY — semaphore was posted (heartbeat arrived);
|
||
* ha_do_demote() was called (role = CTRL_STANDBY).
|
||
* Returns CTRL_ELECTING — sem_timedwait timed out (ETIMEDOUT);
|
||
* caller is responsible for driving the election.
|
||
*/
|
||
ctrl_role_t ha_wait_for_heartbeat_or_timeout(ha_state_t *ha)
|
||
{
|
||
/* Seed jitter PRNG on first use */
|
||
if (ha->jitter_seed == 0)
|
||
ha_seed_jitter(&ha->jitter_seed, ha->my_priority);
|
||
|
||
/* Compute timeout in milliseconds: election_timeout + random jitter */
|
||
long to_ms = (long)ha->config.election_timeout_ms;
|
||
if (ha->config.election_jitter_max_ms > 0) {
|
||
to_ms += (long)(xorshift64_next(&ha->jitter_seed)
|
||
% ((uint64_t)ha->config.election_jitter_max_ms + 1));
|
||
}
|
||
|
||
/* sem_timedwait requires an absolute CLOCK_REALTIME deadline */
|
||
struct timespec deadline;
|
||
clock_gettime(CLOCK_REALTIME, &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 rc = sem_timedwait(&ha->heartbeat_seen, &deadline);
|
||
if (rc == 0) {
|
||
/* Heartbeat arrived before the deadline — demote to standby */
|
||
ha_do_demote(ha);
|
||
return CTRL_STANDBY;
|
||
}
|
||
|
||
/* ETIMEDOUT (or any other error): treat as election timeout */
|
||
return CTRL_ELECTING;
|
||
}
|
||
|
||
/* ── Term Durability (Req 9) ────────────────────────────────────────────────
|
||
*
|
||
* ha_term_file_t: on-disk format for ha-term.bin.
|
||
* Fields are in host byte order (single-host durability, no cross-endian migration).
|
||
* CRC32 covers all preceding bytes via zlib crc32().
|
||
*/
|
||
|
||
#define HA_TERM_MAGIC 0xDEADBEEFCAFEBABEULL
|
||
|
||
typedef struct {
|
||
uint64_t magic;
|
||
uint64_t term;
|
||
uint64_t voted_for_term;
|
||
char voted_for_name[XCE_MAXNAME_LEN];
|
||
uint32_t crc32;
|
||
} ha_term_file_t;
|
||
|
||
/*
|
||
* ha_persist_term: atomically write ha_term_file_t to {workdir}/ha-term.bin.
|
||
* Uses write-to-tmp + fsync + rename for durability (existing project pattern).
|
||
* If workdir is NULL, the persist is skipped (test mode or no workdir configured).
|
||
* Req 9.1: must be called before any vote grant or term increment is reported.
|
||
*/
|
||
void ha_persist_term(ha_state_t *ha, const char *workdir)
|
||
{
|
||
if (!workdir)
|
||
return; /* test mode: skip file I/O */
|
||
|
||
ha_term_file_t f;
|
||
memset(&f, 0, sizeof(f));
|
||
f.magic = HA_TERM_MAGIC;
|
||
f.term = atomic_load_explicit(&ha->term, memory_order_acquire);
|
||
f.voted_for_term = ha->voted_for_term;
|
||
strncpy(f.voted_for_name, ha->voted_for_name, sizeof(f.voted_for_name) - 1);
|
||
f.voted_for_name[sizeof(f.voted_for_name) - 1] = '\0';
|
||
|
||
/* CRC32 covers all bytes before the crc32 field itself */
|
||
size_t crc_len = offsetof(ha_term_file_t, crc32);
|
||
f.crc32 = (uint32_t)crc32(0L, (const Bytef *)&f, (uInt)crc_len);
|
||
|
||
/* Write to tmp then rename — atomic from filesystem perspective */
|
||
char path[1024], tmp[1040];
|
||
snprintf(path, sizeof(path), "%s/ha-term.bin", workdir);
|
||
snprintf(tmp, sizeof(tmp), "%s/ha-term.bin.tmp", workdir);
|
||
|
||
int fd = open(tmp, O_WRONLY | O_CREAT | O_TRUNC, 0600);
|
||
if (fd < 0)
|
||
return; /* best-effort; caller must not rely on this for correctness */
|
||
|
||
ssize_t written = write(fd, &f, sizeof(f));
|
||
if (written == (ssize_t)sizeof(f))
|
||
fsync(fd);
|
||
close(fd);
|
||
|
||
if (written == (ssize_t)sizeof(f))
|
||
rename(tmp, path);
|
||
}
|
||
|
||
/*
|
||
* ha_restore_term: read {workdir}/ha-term.bin at startup and restore
|
||
* ha->term and ha->voted_for_* into ha_state_t.
|
||
*
|
||
* On missing file (ENOENT): no-op — term stays 0. Req 9.3.
|
||
* On CRC/magic mismatch: log WARN, zero ha->term and ha->voted_for_*. Req 9.2.
|
||
* On valid file: atomically restore ha->term; copy voted_for fields. Req 9.2.
|
||
*
|
||
* Returns 0 on success (including gracefully-handled file-not-found and CRC bad),
|
||
* -errno on unexpected I/O error.
|
||
* If workdir is NULL, function is a no-op (safe for unit tests without a workdir).
|
||
*/
|
||
int ha_restore_term(ha_state_t *ha, const char *workdir)
|
||
{
|
||
if (!workdir)
|
||
return 0;
|
||
|
||
char path[1024];
|
||
snprintf(path, sizeof(path), "%s/ha-term.bin", workdir);
|
||
|
||
int fd = open(path, O_RDONLY);
|
||
if (fd < 0) {
|
||
if (errno == ENOENT)
|
||
return 0; /* first boot — no file yet, term stays 0 (Req 9.3) */
|
||
return -errno;
|
||
}
|
||
|
||
ha_term_file_t f;
|
||
memset(&f, 0, sizeof(f));
|
||
ssize_t n = read(fd, &f, sizeof(f));
|
||
close(fd);
|
||
|
||
if (n != (ssize_t)sizeof(f)) {
|
||
WARN("HA: ha-term.bin short read (%zd/%zu) — ignoring, starting at term 0",
|
||
n, sizeof(f));
|
||
atomic_store_explicit(&ha->term, 0ULL, memory_order_release);
|
||
ha->voted_for_term = 0;
|
||
memset(ha->voted_for_name, 0, sizeof(ha->voted_for_name));
|
||
return 0;
|
||
}
|
||
|
||
/* Verify magic */
|
||
if (f.magic != HA_TERM_MAGIC) {
|
||
WARN("HA: ha-term.bin bad magic (0x%016" PRIx64 ") — ignoring, starting at term 0",
|
||
f.magic);
|
||
atomic_store_explicit(&ha->term, 0ULL, memory_order_release);
|
||
ha->voted_for_term = 0;
|
||
memset(ha->voted_for_name, 0, sizeof(ha->voted_for_name));
|
||
return 0;
|
||
}
|
||
|
||
/* Verify CRC32 */
|
||
size_t crc_len = offsetof(ha_term_file_t, crc32);
|
||
uint32_t computed = (uint32_t)crc32(0L, (const Bytef *)&f, (uInt)crc_len);
|
||
if (computed != f.crc32) {
|
||
WARN("HA: ha-term.bin CRC mismatch (expected 0x%08x got 0x%08x) — ignoring, starting at term 0",
|
||
computed, f.crc32);
|
||
atomic_store_explicit(&ha->term, 0ULL, memory_order_release);
|
||
ha->voted_for_term = 0;
|
||
memset(ha->voted_for_name, 0, sizeof(ha->voted_for_name));
|
||
return 0;
|
||
}
|
||
|
||
/* Valid file — restore state (Req 9.2) */
|
||
atomic_store_explicit(&ha->term, f.term, memory_order_release);
|
||
ha->voted_for_term = f.voted_for_term;
|
||
strncpy(ha->voted_for_name, f.voted_for_name, sizeof(ha->voted_for_name) - 1);
|
||
ha->voted_for_name[sizeof(ha->voted_for_name) - 1] = '\0';
|
||
|
||
return 0;
|
||
}
|
||
|
||
/* ── Vote Handler (Req 2.2–2.6, 9.1) ───────────────────────────────────────
|
||
*
|
||
* handle_request_vote: process an incoming OP_requestVote RPC.
|
||
*
|
||
* Decision order (fail-fast):
|
||
* 1. in_catch_up → deny (Req 2.5)
|
||
* 2. req term < my term → deny (Req 2.2)
|
||
* 3. req term > my term → update term, clear voted_for (Req 2.3)
|
||
* 4. already voted for a DIFFERENT candidate in this term → deny (Req 2.6)
|
||
* 5. freshness gate: candidate not at least as fresh → deny (Req 2.4)
|
||
* 6. grant vote, record voted_for, persist (Req 9.1)
|
||
*
|
||
* Returns 1 = granted, 0 = denied, <0 = error.
|
||
*/
|
||
int handle_request_vote(ha_state_t *ha,
|
||
uint64_t term,
|
||
const char *candidate_id,
|
||
uint64_t config_tag,
|
||
uint64_t last_lsn,
|
||
uint64_t my_config_tag,
|
||
uint64_t my_last_lsn,
|
||
int in_catch_up_flag)
|
||
{
|
||
/* Req 2.5: catching-up standbys deny all votes */
|
||
if (in_catch_up_flag)
|
||
return 0;
|
||
|
||
uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire);
|
||
|
||
/* Req 2.2: stale term → deny, do not update any state */
|
||
if (term < my_term)
|
||
return 0;
|
||
|
||
/* Req 2.3: higher term → advance term, clear prior vote record */
|
||
if (term > my_term) {
|
||
atomic_store_explicit(&ha->term, term, memory_order_release);
|
||
ha->voted_for_term = 0;
|
||
memset(ha->voted_for_name, 0, sizeof(ha->voted_for_name));
|
||
}
|
||
|
||
/* Req 2.6: already voted for a different candidate in this term → deny */
|
||
if (ha->voted_for_term == term &&
|
||
ha->voted_for_name[0] != '\0' &&
|
||
strcmp(ha->voted_for_name, candidate_id) != 0)
|
||
return 0;
|
||
|
||
/* Req 2.4: freshness gate
|
||
* Deny if candidate has a lower config_tag, or equal config_tag with lower LSN.
|
||
* Grant if candidate is strictly fresher or equally fresh.
|
||
*/
|
||
if (config_tag < my_config_tag)
|
||
return 0;
|
||
if (config_tag == my_config_tag && last_lsn < my_last_lsn)
|
||
return 0;
|
||
|
||
/* Grant the vote: record voted_for and persist before responding (Req 9.1) */
|
||
ha->voted_for_term = term;
|
||
strncpy(ha->voted_for_name, candidate_id, sizeof(ha->voted_for_name) - 1);
|
||
ha->voted_for_name[sizeof(ha->voted_for_name) - 1] = '\0';
|
||
|
||
/* workdir is not available here — persist with NULL (test mode or caller
|
||
* wraps this function with workdir before sending the network reply).
|
||
* Production dispatch path calls ha_persist_term(ha, workdir) explicitly
|
||
* before this function writes its network reply. */
|
||
ha_persist_term(ha, NULL);
|
||
|
||
return 1;
|
||
}
|
||
|
||
/* ── Election Evaluation (Req 2.1, 2.7, 2.8) ───────────────────────────────
|
||
*
|
||
* ha_elect_evaluate_and_signal: evaluate accumulated votes; set ctx->result
|
||
* and post ctx->done when quorum is reached or all peers replied.
|
||
*
|
||
* Quorum = ceil((n_peers + 1) / 2) where n_peers+1 is total cluster size.
|
||
* ctx->votes_granted starts at 1 (self-vote); ctx->n_peers is n_other_peers.
|
||
* Quorum = ceil((n_peers + 1) / 2) = (n_peers + 1 + 1) / 2 = (n_peers + 2) / 2.
|
||
*
|
||
* Yield gate (Req 2.7): if any responding peer has a higher config_tag, or
|
||
* equal config_tag with a higher LSN than the candidate → yield (CTRL_STANDBY).
|
||
* Candidate's own tag/lsn is stored in ctx->votes[0].my_tag / my_lsn.
|
||
*
|
||
* Fatal gate (Req 2.8): if candidate has config_tag=0 and any peer has tag>0
|
||
* → CTRL_STANDBY (never PRIMARY with no config when peer has config).
|
||
*
|
||
* Returns CTRL_PRIMARY or CTRL_STANDBY; also sets ctx->result.
|
||
*/
|
||
ctrl_role_t ha_elect_evaluate_and_signal(ha_state_t *ha, ha_elect_ctx_t *ctx)
|
||
{
|
||
(void)ha; /* ha->term used for fencing in network path; not needed here */
|
||
|
||
/* Guard: called from handle_request_vote_reply under ctx->lock; called from
|
||
* unit-test path (peers==NULL) without threads — done_posted prevents double
|
||
* sem_post in the network path if two threads both hit quorum simultaneously. */
|
||
if (ctx->done_posted)
|
||
return (ctx->result == 1) ? CTRL_PRIMARY : CTRL_STANDBY;
|
||
|
||
int total_nodes = ctx->n_peers + 1; /* include self */
|
||
int quorum = (total_nodes + 1) / 2; /* ceil(total/2) */
|
||
|
||
/* Req 2.8: fatal gate — candidate has no config but a peer does */
|
||
uint64_t my_tag = ctx->votes[0].my_tag;
|
||
uint64_t my_lsn = ctx->votes[0].my_lsn;
|
||
if (my_tag == 0 && ctx->max_resp_tag > 0) {
|
||
ctx->result = 0;
|
||
ctx->done_posted = 1;
|
||
sem_post(&ctx->done);
|
||
return CTRL_STANDBY;
|
||
}
|
||
|
||
/* Check quorum */
|
||
if (ctx->votes_granted < quorum) {
|
||
ctx->result = 0;
|
||
ctx->done_posted = 1;
|
||
sem_post(&ctx->done);
|
||
return CTRL_STANDBY;
|
||
}
|
||
|
||
/* Req 2.7: yield gate — fresher responding peer */
|
||
if (ctx->max_resp_tag > my_tag) {
|
||
ctx->result = 0;
|
||
ctx->done_posted = 1;
|
||
sem_post(&ctx->done);
|
||
return CTRL_STANDBY;
|
||
}
|
||
if (ctx->max_resp_tag == my_tag && ctx->max_resp_lsn > my_lsn) {
|
||
ctx->result = 0;
|
||
ctx->done_posted = 1;
|
||
sem_post(&ctx->done);
|
||
return CTRL_STANDBY;
|
||
}
|
||
|
||
/* Quorum met and no fresher peer → PRIMARY */
|
||
ctx->result = 1;
|
||
ctx->done_posted = 1;
|
||
sem_post(&ctx->done);
|
||
return CTRL_PRIMARY;
|
||
}
|
||
|
||
/* ── Vote Reply Handler (Req 2.2–2.6) ──────────────────────────────────────
|
||
*
|
||
* handle_request_vote_reply: process a reply to an OP_requestVote RPC.
|
||
*
|
||
* Called by each per-peer vote-request thread once the peer replies.
|
||
* Updates ctx->votes[peer_idx] with the reply fields, tracks the freshest
|
||
* responding peer in ctx->max_resp_tag/max_resp_lsn, increments counters,
|
||
* and evaluates whether to signal ctx->done.
|
||
*
|
||
* Quorum = ceil((n_peers + 1) / 2) = (n_peers + 2) / 2
|
||
* (n_peers = number of other peers, not counting self; votes_granted starts at 1)
|
||
*
|
||
* Posts ctx->done via ha_elect_evaluate_and_signal when:
|
||
* - votes_granted reaches quorum (fast path — no need to wait for all replies), OR
|
||
* - n_replied reaches n_peers (all peers accounted for, win or lose)
|
||
*/
|
||
void handle_request_vote_reply(ha_state_t *ha,
|
||
ha_elect_ctx_t *ctx,
|
||
int peer_idx,
|
||
uint64_t term,
|
||
int vote_granted,
|
||
uint64_t peer_config_tag,
|
||
uint64_t peer_lsn)
|
||
{
|
||
(void)term; /* fencing: callers bump ha->term if peer's term > ours, before calling here */
|
||
|
||
/* Guard against out-of-range index */
|
||
if (peer_idx < 0 || peer_idx >= ctx->n_peers)
|
||
return;
|
||
|
||
pthread_mutex_lock(&ctx->lock);
|
||
|
||
ha_vote_entry_t *entry = &ctx->votes[peer_idx];
|
||
|
||
/* Idempotency guard: ignore duplicate replies */
|
||
if (entry->replied) {
|
||
pthread_mutex_unlock(&ctx->lock);
|
||
return;
|
||
}
|
||
|
||
/* Record reply fields */
|
||
entry->replied = 1;
|
||
entry->vote_granted = vote_granted ? 1 : 0;
|
||
entry->peer_tag = peer_config_tag;
|
||
entry->peer_lsn = peer_lsn;
|
||
|
||
/* Update counters */
|
||
if (vote_granted)
|
||
ctx->votes_granted++;
|
||
ctx->n_replied++;
|
||
|
||
/* Track the freshest responding peer for the yield gate (Req 2.7) */
|
||
if (peer_config_tag > ctx->max_resp_tag) {
|
||
ctx->max_resp_tag = peer_config_tag;
|
||
ctx->max_resp_lsn = peer_lsn;
|
||
} else if (peer_config_tag == ctx->max_resp_tag && peer_lsn > ctx->max_resp_lsn) {
|
||
ctx->max_resp_lsn = peer_lsn;
|
||
}
|
||
|
||
/* Decide whether to signal the waiting thread.
|
||
* Quorum = ceil((n_peers + 1) / 2) where n_peers+1 is total cluster size.
|
||
* n_peers is the peer count excluding self; self-vote is already in votes_granted.
|
||
* ha_elect_evaluate_and_signal is called under ctx->lock; its done_posted guard
|
||
* prevents double sem_post if two threads reach quorum simultaneously. */
|
||
int total_nodes = ctx->n_peers + 1;
|
||
int quorum = (total_nodes + 1) / 2;
|
||
|
||
if (ctx->votes_granted >= quorum || ctx->n_replied >= ctx->n_peers)
|
||
ha_elect_evaluate_and_signal(ha, ctx);
|
||
|
||
pthread_mutex_unlock(&ctx->lock);
|
||
}
|
||
|
||
/* ── Per-peer vote-request thread ──────────────────────────────────────────
|
||
*
|
||
* vote_thread_ctx_t: arguments passed to each vote_request_thread().
|
||
* One struct per peer; allocated on the stack of am_i_primary_quorum.
|
||
*/
|
||
typedef struct {
|
||
ha_state_t *ha; /* Candidate's HA state (term, ssl_ctx) */
|
||
ha_elect_ctx_t *ctx; /* Shared election context */
|
||
xce_candidate_t *peer; /* Target peer */
|
||
int peer_idx; /* Index in ctx->votes[] */
|
||
uint64_t my_term; /* Term sent in the request */
|
||
uint64_t my_tag; /* config_tag sent in the request */
|
||
uint64_t my_lsn; /* WAL LSN sent in the request */
|
||
int timeout_ms; /* Connect + I/O timeout */
|
||
} vote_thread_ctx_t;
|
||
|
||
/*
|
||
* vote_request_thread: per-peer pthread that sends OP_requestVote and waits
|
||
* for a RequestVoteReply.
|
||
*
|
||
* Wire format (raw SSL, no iomgr):
|
||
* Send: uint64_t size_hdr + capnp(RequestVoteRequest)
|
||
* Receive: uint64_t size_hdr + capnp(RequestVoteReply)
|
||
*
|
||
* On any error (connect, encode, I/O, decode) the peer counts as denied with
|
||
* zero peer_tag/peer_lsn — the quorum evaluator sees an absent peer as a no-vote.
|
||
*/
|
||
static void *vote_request_thread(void *arg)
|
||
{
|
||
vote_thread_ctx_t *vctx = (vote_thread_ctx_t *)arg;
|
||
ha_state_t *ha = vctx->ha;
|
||
ha_elect_ctx_t *ctx = vctx->ctx;
|
||
int idx = vctx->peer_idx;
|
||
|
||
SSL *ssl = NULL;
|
||
int sock = -1;
|
||
|
||
/* ── 1. TCP connect (non-blocking poll loop) ─────────── */
|
||
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), "%d", vctx->peer->port);
|
||
if (getaddrinfo(vctx->peer->hostname, port_str, &hints, &res) != 0)
|
||
goto deny;
|
||
|
||
sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
|
||
if (sock < 0) { freeaddrinfo(res); sock = -1; goto deny; }
|
||
|
||
{
|
||
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;
|
||
|
||
struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP };
|
||
int pr;
|
||
do { pr = poll(&pfd, 1, vctx->timeout_ms); } while (pr < 0 && errno == EINTR);
|
||
if (pr <= 0 || (pfd.revents & (POLLERR | POLLHUP))) goto deny;
|
||
|
||
int err = 0; socklen_t el = sizeof(err);
|
||
getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
|
||
if (err != 0) goto deny;
|
||
|
||
/* Restore blocking for SSL handshake */
|
||
fcntl(sock, F_SETFL, flags & ~O_NONBLOCK);
|
||
}
|
||
|
||
/* ── 2. SSL handshake ────────────────────────────────── */
|
||
if (ha->ssl_ctx) {
|
||
ssl = SSL_new((SSL_CTX *)ha->ssl_ctx);
|
||
if (!ssl) goto deny;
|
||
SSL_set_fd(ssl, sock);
|
||
int sr;
|
||
do {
|
||
sr = SSL_connect(ssl);
|
||
if (sr <= 0) {
|
||
int e = SSL_get_error(ssl, sr);
|
||
if (e != SSL_ERROR_WANT_READ && e != SSL_ERROR_WANT_WRITE) goto deny;
|
||
}
|
||
} while (sr <= 0);
|
||
if (sr <= 0) goto deny;
|
||
}
|
||
|
||
/* ── 3. Encode + send RequestVoteRequest ─────────────── */
|
||
{
|
||
struct capn c;
|
||
struct capn_segment *cs;
|
||
struct RequestVoteRequest req_s;
|
||
RequestVoteRequest_ptr req_p;
|
||
uint8_t *buf = NULL;
|
||
int64_t blen = 0;
|
||
|
||
capn_init_malloc(&c);
|
||
cs = capn_root(&c).seg;
|
||
memset(&req_s, 0, sizeof(req_s));
|
||
req_s.term = vctx->my_term;
|
||
req_s.candidateId.str = ""; req_s.candidateId.len = 0; /* no name needed here */
|
||
req_s.configTag = vctx->my_tag;
|
||
req_s.lastLogLsn = vctx->my_lsn;
|
||
|
||
req_p = new_RequestVoteRequest(cs);
|
||
write_RequestVoteRequest(&req_s, req_p);
|
||
capn_setp(capn_root(&c), 0, req_p.p);
|
||
|
||
blen = capn_size(&c);
|
||
if (blen > 0) buf = malloc((size_t)blen);
|
||
if (!buf || capn_write_mem(&c, buf, (size_t)blen, 0) < 0) {
|
||
free(buf); capn_free(&c); goto deny;
|
||
}
|
||
capn_free(&c);
|
||
|
||
uint64_t hdr = (uint64_t)blen;
|
||
int nsent;
|
||
if (ssl) {
|
||
nsent = SSL_write(ssl, &hdr, (int)sizeof(hdr));
|
||
if (nsent == (int)sizeof(hdr))
|
||
nsent = SSL_write(ssl, buf, (int)blen);
|
||
} else {
|
||
nsent = (int)write(sock, &hdr, sizeof(hdr));
|
||
if (nsent == (int)sizeof(hdr))
|
||
nsent = (int)write(sock, buf, (size_t)blen);
|
||
}
|
||
free(buf);
|
||
if (nsent != (int)blen) goto deny;
|
||
}
|
||
|
||
/* ── 4. Read + decode RequestVoteReply ───────────────── */
|
||
{
|
||
uint64_t hdr = 0;
|
||
size_t off = 0;
|
||
|
||
/* Read the 8-byte size prefix */
|
||
while (off < sizeof(hdr)) {
|
||
int n;
|
||
if (ssl)
|
||
n = SSL_read(ssl, (char *)&hdr + off, (int)(sizeof(hdr) - off));
|
||
else
|
||
n = (int)read(sock, (char *)&hdr + off, sizeof(hdr) - off);
|
||
if (n <= 0) goto deny;
|
||
off += (size_t)n;
|
||
}
|
||
|
||
if (hdr == 0 || hdr > 65536) goto deny; /* sanity cap */
|
||
|
||
uint8_t *buf = malloc((size_t)hdr);
|
||
if (!buf) goto deny;
|
||
|
||
off = 0;
|
||
while (off < (size_t)hdr) {
|
||
int n;
|
||
if (ssl)
|
||
n = SSL_read(ssl, (char *)buf + off, (int)((size_t)hdr - off));
|
||
else
|
||
n = (int)read(sock, (char *)buf + off, (size_t)hdr - off);
|
||
if (n <= 0) { free(buf); goto deny; }
|
||
off += (size_t)n;
|
||
}
|
||
|
||
struct capn c;
|
||
struct RequestVoteReply rep_s;
|
||
RequestVoteReply_ptr rep_p;
|
||
|
||
if (capn_init_mem(&c, buf, (size_t)hdr, 0) != 0) {
|
||
free(buf); goto deny;
|
||
}
|
||
rep_p.p = capn_getp(capn_root(&c), 0, 1);
|
||
read_RequestVoteReply(&rep_s, rep_p);
|
||
capn_free(&c);
|
||
free(buf);
|
||
|
||
/* Fencing: if peer's term > ours, step down immediately */
|
||
if (rep_s.term > vctx->my_term)
|
||
atomic_store_explicit(&ha->term, rep_s.term, memory_order_release);
|
||
|
||
/* Record the reply */
|
||
handle_request_vote_reply(ha, ctx, idx,
|
||
rep_s.term,
|
||
rep_s.voteGranted ? 1 : 0,
|
||
rep_s.peerConfigTag,
|
||
rep_s.peerLastLogLsn);
|
||
}
|
||
|
||
goto cleanup;
|
||
|
||
deny:
|
||
/* Treat unreachable / error peer as denied vote with zero freshness */
|
||
handle_request_vote_reply(ha, ctx, idx,
|
||
vctx->my_term, /* use our term so no fencing */
|
||
0, /* vote_granted = false */
|
||
0, 0);
|
||
cleanup:
|
||
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
|
||
if (sock >= 0) close(sock);
|
||
return NULL;
|
||
}
|
||
|
||
/* ── Top-Level Election Entrypoint (Req 2.1, 2.9, 2.10, 2.11) ──────────────
|
||
*
|
||
* am_i_primary_quorum: top-level quorum election.
|
||
*
|
||
* For single-node clusters (n_peers == 1): self-elect immediately (Req 2.9).
|
||
* For multi-node clusters: increment term, allocate ha_elect_ctx_t, spawn
|
||
* per-peer pthreads to send OP_requestVote, block on ctx->done, evaluate result.
|
||
* Split vote → one retry with incremented term and randomized jitter (Req 2.10).
|
||
* Second failure → CTRL_STANDBY (Req 2.11).
|
||
*
|
||
* Unit tests: peers==NULL and n_peers==1 exercises the single-node shortcircuit.
|
||
* Multi-peer network path (peers!=NULL) is exercised by integration tests.
|
||
*/
|
||
ctrl_role_t am_i_primary_quorum(ha_state_t *ha,
|
||
int n_peers,
|
||
xce_candidate_t **peers,
|
||
uint64_t my_config_tag,
|
||
uint64_t my_lsn)
|
||
{
|
||
/* Req 2.9: single-node self-elects immediately */
|
||
if (n_peers == 1) {
|
||
atomic_fetch_add_explicit(&ha->term, 1, memory_order_acq_rel);
|
||
uint64_t term = atomic_load_explicit(&ha->term, memory_order_acquire);
|
||
INFO("HA election: outcome=PRIMARY term=%" PRIu64
|
||
" votes_granted=1 quorum=1 (single-node)", term);
|
||
INFO("HA FSM: CTRL_INIT → CTRL_PRIMARY (single-node, term=%" PRIu64 ")", term);
|
||
return CTRL_PRIMARY;
|
||
}
|
||
|
||
/* Multi-node path: up to 2 election rounds (Req 2.10, 2.11) */
|
||
int n_other = n_peers - 1; /* number of peers excluding self */
|
||
int total_nodes = n_peers;
|
||
int quorum = (total_nodes + 1) / 2;
|
||
|
||
for (int round = 0; round < 2; round++) {
|
||
/* Increment term before each election round */
|
||
atomic_fetch_add_explicit(&ha->term, 1, memory_order_acq_rel);
|
||
uint64_t cur_term = atomic_load_explicit(&ha->term, memory_order_acquire);
|
||
|
||
/* Req 10.5: log FSM transition to CTRL_ELECTING */
|
||
INFO("HA FSM: → CTRL_ELECTING (round=%d, term=%" PRIu64 ", n_peers=%d)",
|
||
round + 1, cur_term, n_peers);
|
||
atomic_store_explicit(&ha->role, (int32_t)CTRL_ELECTING, memory_order_release);
|
||
|
||
/* Add jitter delay between rounds (Req 2.10) */
|
||
if (round > 0 && ha->config.election_jitter_max_ms > 0) {
|
||
if (ha->jitter_seed == 0)
|
||
ha_seed_jitter(&ha->jitter_seed, ha->my_priority);
|
||
long jitter_ms = (long)(xorshift64_next(&ha->jitter_seed)
|
||
% ((uint64_t)ha->config.election_jitter_max_ms + 1));
|
||
if (jitter_ms > 0) {
|
||
struct timespec ts = {
|
||
.tv_sec = jitter_ms / 1000,
|
||
.tv_nsec = (jitter_ms % 1000) * 1000000L,
|
||
};
|
||
nanosleep(&ts, NULL);
|
||
}
|
||
}
|
||
|
||
/* Allocate election context */
|
||
ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) +
|
||
(size_t)n_other * sizeof(ctx->votes[0]));
|
||
if (!ctx)
|
||
return CTRL_STANDBY;
|
||
|
||
ctx->n_peers = n_other;
|
||
ctx->votes_granted = 1; /* self-vote */
|
||
ctx->n_replied = 0;
|
||
ctx->max_resp_tag = 0;
|
||
ctx->max_resp_lsn = 0;
|
||
ctx->result = -1;
|
||
ctx->done_posted = 0;
|
||
pthread_mutex_init(&ctx->lock, NULL);
|
||
sem_init(&ctx->done, 0, 0);
|
||
|
||
/* Store candidate's own tag/lsn in votes[0] for evaluate function */
|
||
if (n_other > 0) {
|
||
ctx->votes[0].my_tag = my_config_tag;
|
||
ctx->votes[0].my_lsn = my_lsn;
|
||
}
|
||
|
||
/* Spawn per-peer vote-request threads when peers list is provided.
|
||
* When peers==NULL (unit test mode), skip spawning and evaluate
|
||
* immediately with only the self-vote. */
|
||
vote_thread_ctx_t *vctxs = NULL;
|
||
pthread_t *tids = NULL;
|
||
|
||
if (peers != NULL && n_other > 0) {
|
||
vctxs = calloc((size_t)n_other, sizeof(*vctxs));
|
||
tids = calloc((size_t)n_other, sizeof(*tids));
|
||
if (!vctxs || !tids) {
|
||
free(vctxs); free(tids);
|
||
pthread_mutex_destroy(&ctx->lock);
|
||
sem_destroy(&ctx->done);
|
||
free(ctx);
|
||
return CTRL_STANDBY;
|
||
}
|
||
|
||
for (int i = 0; i < n_other; i++) {
|
||
vctxs[i].ha = ha;
|
||
vctxs[i].ctx = ctx;
|
||
vctxs[i].peer = peers[i];
|
||
vctxs[i].peer_idx = i;
|
||
vctxs[i].my_term = cur_term;
|
||
vctxs[i].my_tag = my_config_tag;
|
||
vctxs[i].my_lsn = my_lsn;
|
||
vctxs[i].timeout_ms =
|
||
(int)ha->config.probe_connect_timeout_ms;
|
||
pthread_create(&tids[i], NULL, vote_request_thread, &vctxs[i]);
|
||
}
|
||
|
||
/* Block until quorum or all peers replied */
|
||
sem_wait(&ctx->done);
|
||
|
||
/* Join all threads before freeing ctx */
|
||
for (int i = 0; i < n_other; i++)
|
||
pthread_join(tids[i], NULL);
|
||
|
||
free(vctxs);
|
||
free(tids);
|
||
}
|
||
|
||
ctrl_role_t role;
|
||
if (peers == NULL) {
|
||
/* Unit-test path: no network peers; evaluate with self-vote only */
|
||
role = ha_elect_evaluate_and_signal(ha, ctx);
|
||
} else {
|
||
/* Production path: ctx->result was set by ha_elect_evaluate_and_signal
|
||
* inside handle_request_vote_reply; just read it back. */
|
||
role = (ctx->result == 1) ? CTRL_PRIMARY : CTRL_STANDBY;
|
||
}
|
||
|
||
/* Capture before freeing ctx */
|
||
int votes_granted_snap = ctx->votes_granted;
|
||
|
||
/* Req 10.1: log election outcome */
|
||
INFO("HA election: outcome=%s term=%" PRIu64
|
||
" votes_granted=%d quorum=%d (round=%d)",
|
||
(role == CTRL_PRIMARY) ? "PRIMARY" : "STANDBY",
|
||
cur_term, votes_granted_snap, quorum, round + 1);
|
||
|
||
/* Req 10.5: FSM log for multi-node CTRL_PRIMARY path.
|
||
* ha_do_promote() is not called here so its FSM log is absent; emit it now. */
|
||
if (role == CTRL_PRIMARY)
|
||
INFO("HA FSM: → CTRL_PRIMARY (term=%" PRIu64 " round=%d votes=%d quorum=%d)",
|
||
cur_term, round + 1, votes_granted_snap, quorum);
|
||
|
||
pthread_mutex_destroy(&ctx->lock);
|
||
sem_destroy(&ctx->done);
|
||
free(ctx);
|
||
|
||
if (role == CTRL_PRIMARY)
|
||
return CTRL_PRIMARY;
|
||
|
||
/* Split vote: retry once (Req 2.10); second failure → CTRL_STANDBY (Req 2.11) */
|
||
}
|
||
|
||
return CTRL_STANDBY;
|
||
}
|
||
|
||
/* ── Task 9.2: Persistent Control Channel ──────────────────────────────────
|
||
*
|
||
* ha_find_and_connect_to_primary: three-tier primary discovery + TCP connect.
|
||
*
|
||
* Requirements: 6.1
|
||
* Boundary: Persistent Control Channel (ha.c)
|
||
*/
|
||
|
||
#ifdef XCE_TESTING
|
||
int (*ha_tcp_connect_hook)(const char *host, int port) = NULL;
|
||
#endif
|
||
|
||
/* ha_tcp_connect_raw: non-blocking connect + poll, returns connected fd or -errno */
|
||
static int ha_tcp_connect_raw(const char *host, int port, int timeout_ms)
|
||
{
|
||
#ifdef XCE_TESTING
|
||
if (ha_tcp_connect_hook)
|
||
return ha_tcp_connect_hook(host, port);
|
||
#endif
|
||
|
||
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), "%d", port);
|
||
if (getaddrinfo(host, port_str, &hints, &res) != 0)
|
||
return -EHOSTUNREACH;
|
||
|
||
int sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
|
||
if (sock < 0) { int e = errno; freeaddrinfo(res); return -e; }
|
||
|
||
int one = 1;
|
||
setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
|
||
|
||
int flags = fcntl(sock, F_GETFL, 0);
|
||
fcntl(sock, F_SETFL, flags | O_NONBLOCK);
|
||
connect(sock, res->ai_addr, res->ai_addrlen);
|
||
freeaddrinfo(res);
|
||
|
||
struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP };
|
||
int pr;
|
||
do { pr = poll(&pfd, 1, timeout_ms); } while (pr < 0 && errno == EINTR);
|
||
|
||
if (pr == 0) { close(sock); return -ETIMEDOUT; }
|
||
if (pr < 0 || (pfd.revents & (POLLERR | POLLHUP))) {
|
||
close(sock); return -ECONNREFUSED;
|
||
}
|
||
int err = 0; socklen_t el = sizeof(err);
|
||
getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
|
||
if (err) { close(sock); return -err; }
|
||
|
||
fcntl(sock, F_SETFL, flags & ~O_NONBLOCK);
|
||
return sock;
|
||
}
|
||
|
||
static int find_candidate_by_name(const xce_cluster_config_t *peers,
|
||
const char *name)
|
||
{
|
||
for (int i = 0; i < peers->candidates_count; i++) {
|
||
if (strcmp(peers->candidates[i].name, name) == 0)
|
||
return i;
|
||
}
|
||
return -1;
|
||
}
|
||
|
||
static void register_fd_with_iomgr(cluster_t *c, int fd)
|
||
{
|
||
if (!c) return;
|
||
struct epoll_event ev;
|
||
memset(&ev, 0, sizeof(ev));
|
||
ev.events = EPOLLIN | EPOLLRDHUP;
|
||
ev.data.fd = fd;
|
||
if (epoll_ctl(c->iomgr.epoll_fd, EPOLL_CTL_ADD, fd, &ev) != 0)
|
||
WARN("ha_find_and_connect: epoll_ctl: %s", strerror(errno));
|
||
}
|
||
|
||
static void ms_sleep_ha(long delay_ms)
|
||
{
|
||
if (delay_ms <= 0) return;
|
||
struct timespec ts = { .tv_sec = delay_ms / 1000,
|
||
.tv_nsec = (delay_ms % 1000) * 1000000L };
|
||
nanosleep(&ts, NULL);
|
||
}
|
||
|
||
int ha_find_and_connect_to_primary(ha_state_t *ha,
|
||
const ha_config_t *cfg,
|
||
const xce_cluster_config_t *peers,
|
||
cluster_t *c)
|
||
{
|
||
if (!ha || !cfg || !peers || peers->candidates_count <= 0)
|
||
return -EINVAL;
|
||
|
||
int timeout_ms = (int)cfg->probe_connect_timeout_ms;
|
||
if (timeout_ms <= 0)
|
||
timeout_ms = HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS;
|
||
|
||
/* Tier 1: voted_for_name → direct connect */
|
||
if (ha->voted_for_name[0] != '\0') {
|
||
int idx = find_candidate_by_name(peers, ha->voted_for_name);
|
||
if (idx >= 0) {
|
||
const xce_candidate_t *p1 = &peers->candidates[idx];
|
||
for (int r = 0; r < HA_FIND_PRIMARY_T1_MAX_RETRIES; r++) {
|
||
int fd = ha_tcp_connect_raw(p1->hostname, p1->port, timeout_ms);
|
||
if (fd >= 0) {
|
||
INFO("HA: Tier 1 connected to voted_for '%s' (retry=%d)",
|
||
ha->voted_for_name, r);
|
||
ha->primary_ctrl_fd = fd;
|
||
register_fd_with_iomgr(c, fd);
|
||
return fd;
|
||
}
|
||
if (fd == -ETIMEDOUT || fd == -EAGAIN) {
|
||
WARN("HA: Tier 1 '%s' not ready (retry %d/%d), backoff %dms",
|
||
ha->voted_for_name, r + 1,
|
||
HA_FIND_PRIMARY_T1_MAX_RETRIES, HA_FIND_PRIMARY_T1_BACKOFF_MS);
|
||
ms_sleep_ha(HA_FIND_PRIMARY_T1_BACKOFF_MS);
|
||
continue;
|
||
}
|
||
break; /* other errors → Tier 2 */
|
||
}
|
||
}
|
||
}
|
||
|
||
/* Tier 2: probe remaining candidates */
|
||
{
|
||
int t1_idx = find_candidate_by_name(peers, ha->voted_for_name);
|
||
for (int i = 0; i < peers->candidates_count; i++) {
|
||
if (i == t1_idx) continue;
|
||
const xce_candidate_t *p2 = &peers->candidates[i];
|
||
int fd = ha_tcp_connect_raw(p2->hostname, p2->port, timeout_ms);
|
||
if (fd >= 0) {
|
||
INFO("HA: Tier 2 connected to '%s'", p2->name);
|
||
ha->primary_ctrl_fd = fd;
|
||
register_fd_with_iomgr(c, fd);
|
||
return fd;
|
||
}
|
||
}
|
||
}
|
||
|
||
/* Tier 3: exponential backoff scan (no exit) */
|
||
INFO("HA: entering Tier 3 primary discovery");
|
||
long backoff_ms = HA_FIND_PRIMARY_T3_INIT_MS;
|
||
for (;;) {
|
||
for (int i = 0; i < peers->candidates_count; i++) {
|
||
const xce_candidate_t *p3 = &peers->candidates[i];
|
||
int fd = ha_tcp_connect_raw(p3->hostname, p3->port, timeout_ms);
|
||
if (fd >= 0) {
|
||
INFO("HA: Tier 3 connected to '%s' after backoff", p3->name);
|
||
ha->primary_ctrl_fd = fd;
|
||
register_fd_with_iomgr(c, fd);
|
||
return fd;
|
||
}
|
||
}
|
||
WARN("HA: Tier 3 — no primary, sleep %ldms", backoff_ms);
|
||
ms_sleep_ha(backoff_ms);
|
||
backoff_ms = backoff_ms * 2 > HA_FIND_PRIMARY_T3_MAX_MS
|
||
? HA_FIND_PRIMARY_T3_MAX_MS : backoff_ms * 2;
|
||
}
|
||
}
|
||
|
||
/* ── Task 9.4: EPOLLRDHUP Failover and Heartbeat Timeout ───────────────────
|
||
*
|
||
* Requirements: 6.2 (RDHUP fast path), 6.3 (heartbeat timeout fallback),
|
||
* 6.4 (no heartbeat within timeout → elect), 6.8 (role flip)
|
||
* 10.5 (INFO logs on FSM transitions)
|
||
*/
|
||
|
||
void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c)
|
||
{
|
||
if (!ha) return;
|
||
|
||
/* Close and null the primary control fd */
|
||
if (ha->primary_ctrl_fd >= 0) {
|
||
close(ha->primary_ctrl_fd);
|
||
ha->primary_ctrl_fd = -1;
|
||
}
|
||
|
||
/* Transition to CTRL_ELECTING (Req 6.2) */
|
||
atomic_store_explicit(&ha->role, CTRL_ELECTING, memory_order_release);
|
||
INFO("HA: primary_ctrl_fd RDHUP → CTRL_ELECTING (Req 10.5)");
|
||
|
||
if (!c) return; /* unit test mode: skip persist + election */
|
||
|
||
/* Truncate uncommitted WAL entries above committed_lsn (Req 6.2) */
|
||
uint64_t committed = atomic_load_explicit(&ha->last_known_committed_lsn,
|
||
memory_order_acquire);
|
||
if (committed > 0)
|
||
persist_truncate_wal_above(&c->persist, committed);
|
||
|
||
/* Fire election — am_i_primary_quorum handles the vote and role assignment */
|
||
if (c->myconf && c->myconf->peers_cfg.candidates_count > 0) {
|
||
const xce_cluster_config_t *cfg = &c->myconf->peers_cfg;
|
||
int n = cfg->candidates_count > 0 ? cfg->candidates_count - 1 : 0;
|
||
/* Build peer pointer array (exclude self) */
|
||
xce_candidate_t *peer_ptrs[MAX_HA_CANDIDATES];
|
||
int np = 0;
|
||
for (int i = 0; i < cfg->candidates_count && np < n; i++) {
|
||
if (strcmp(cfg->candidates[i].name, c->myhost ? c->myhost->name : "") != 0)
|
||
peer_ptrs[np++] = &cfg->candidates[i];
|
||
}
|
||
uint64_t my_tag = c->myconf->config_version_tag;
|
||
uint64_t my_lsn = atomic_load_explicit(&c->persist.committed_lsn,
|
||
memory_order_acquire);
|
||
ctrl_role_t new_role = am_i_primary_quorum(ha, np, peer_ptrs, my_tag, my_lsn);
|
||
if (new_role == CTRL_PRIMARY)
|
||
INFO("HA: RDHUP recovery → CTRL_PRIMARY (Req 6.8)");
|
||
else
|
||
INFO("HA: RDHUP recovery → CTRL_STANDBY, reconnecting (Req 6.8)");
|
||
}
|
||
}
|
||
|
||
/* ── Heartbeat timeout watchdog ─────────────────────────────────────────── */
|
||
|
||
typedef struct {
|
||
ha_state_t *ha;
|
||
cluster_t *c;
|
||
sem_t reset_sem; /* ha_on_heartbeat_received posts here */
|
||
atomic_int disarmed; /* set to 1 by ha_disarm_heartbeat_timeout_vtask */
|
||
pthread_t thread;
|
||
int started;
|
||
} hb_vtask_t;
|
||
|
||
/* File-scope pointer to the active heartbeat watchdog.
|
||
* Safety: arm/disarm/heartbeat are called only from the single-threaded
|
||
* vtaskmgr dispatch context; no locking is needed in the current architecture.
|
||
* If concurrent access is added later, protect with ha->lock. */
|
||
static hb_vtask_t *s_hb_vtask = NULL;
|
||
|
||
static void *hb_timeout_thread(void *arg)
|
||
{
|
||
hb_vtask_t *t = (hb_vtask_t *)arg;
|
||
ha_state_t *ha = t->ha;
|
||
|
||
while (!atomic_load(&t->disarmed)) {
|
||
/* Compute deadline: election_timeout_ms + rand_jitter */
|
||
uint32_t jitter_ms = 0;
|
||
if (ha->config.election_jitter_max_ms > 0) {
|
||
struct timespec now;
|
||
clock_gettime(CLOCK_REALTIME, &now);
|
||
unsigned int seed = (unsigned int)now.tv_nsec;
|
||
jitter_ms = (uint32_t)(rand_r(&seed) % ha->config.election_jitter_max_ms);
|
||
}
|
||
uint32_t total_ms = ha->config.election_timeout_ms + jitter_ms;
|
||
|
||
struct timespec deadline;
|
||
clock_gettime(CLOCK_REALTIME, &deadline);
|
||
deadline.tv_sec += (long)(total_ms / 1000);
|
||
deadline.tv_nsec += (long)(total_ms % 1000) * 1000000L;
|
||
if (deadline.tv_nsec >= 1000000000L) {
|
||
deadline.tv_sec++;
|
||
deadline.tv_nsec -= 1000000000L;
|
||
}
|
||
|
||
/* Wait for heartbeat signal or deadline */
|
||
int rc = sem_timedwait(&t->reset_sem, &deadline);
|
||
if (rc == -1 && errno == ETIMEDOUT) {
|
||
/* Timeout with no heartbeat → failover (Req 6.3, 6.4) */
|
||
if (!atomic_load(&t->disarmed)) {
|
||
WARN("HA: heartbeat timeout (%ums) → EPOLLRDHUP failover path",
|
||
total_ms);
|
||
ha_on_primary_ctrl_fd_rdhup(ha, t->c);
|
||
break; /* fired; thread exits */
|
||
}
|
||
}
|
||
/* heartbeat received — loop and re-arm deadline */
|
||
}
|
||
return NULL;
|
||
}
|
||
|
||
void ha_arm_heartbeat_timeout_vtask(ha_state_t *ha, cluster_t *c)
|
||
{
|
||
if (!ha) return;
|
||
|
||
/* Disarm any existing watchdog */
|
||
if (s_hb_vtask && s_hb_vtask->started) {
|
||
atomic_store(&s_hb_vtask->disarmed, 1);
|
||
sem_post(&s_hb_vtask->reset_sem); /* wake sleeping thread */
|
||
pthread_join(s_hb_vtask->thread, NULL);
|
||
sem_destroy(&s_hb_vtask->reset_sem);
|
||
free(s_hb_vtask);
|
||
s_hb_vtask = NULL;
|
||
}
|
||
|
||
hb_vtask_t *t = calloc(1, sizeof(*t));
|
||
if (!t) return;
|
||
|
||
t->ha = ha;
|
||
t->c = c;
|
||
t->started = 0;
|
||
atomic_init(&t->disarmed, 0);
|
||
sem_init(&t->reset_sem, 0, 0);
|
||
|
||
if (pthread_create(&t->thread, NULL, hb_timeout_thread, t) != 0) {
|
||
sem_destroy(&t->reset_sem);
|
||
free(t);
|
||
return;
|
||
}
|
||
t->started = 1;
|
||
s_hb_vtask = t;
|
||
}
|
||
|
||
void ha_on_heartbeat_received(ha_state_t *ha)
|
||
{
|
||
(void)ha;
|
||
if (s_hb_vtask && s_hb_vtask->started)
|
||
sem_post(&s_hb_vtask->reset_sem);
|
||
}
|
||
|
||
void ha_disarm_heartbeat_timeout_vtask(ha_state_t *ha)
|
||
{
|
||
(void)ha;
|
||
if (!s_hb_vtask) return;
|
||
if (s_hb_vtask->started) {
|
||
atomic_store(&s_hb_vtask->disarmed, 1);
|
||
sem_post(&s_hb_vtask->reset_sem);
|
||
pthread_join(s_hb_vtask->thread, NULL);
|
||
sem_destroy(&s_hb_vtask->reset_sem);
|
||
}
|
||
free(s_hb_vtask);
|
||
s_hb_vtask = NULL;
|
||
}
|