wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity
myxce/src/controller/ha.c
Rongsong Shen 72776f18ff feat(ha-config-bootstrap): implement dispatch gate (task 3.3) 
- errnox.h: add E_XCE_NOT_READY and E_XCE_STANDBY error codes with
  extern int XCE_ERR_NOT_READY / XCE_ERR_STANDBY runtime variables
- ha.h: declare ha_dispatch_gate(ha_state_t *, int opcode)
- ha.c: define XCE_ERR_* variables; initialize heartbeat_seen semaphore
  and primary_ctrl_fd = -1 in ha_init(); sem_destroy in ha_fini();
  implement ha_dispatch_gate() with allowlist for CTRL_INIT/CTRL_ELECTING
  (only OP_requestVote and OP_probeConfigTag pass) and CTRL_STANDBY
  (replication, election, and housekeeping ops pass; primary-only ops
  return -XCE_ERR_STANDBY)
- cluster.c: wire ha_dispatch_gate() in myreq_handler() before opcode
  switch; gate failure sends error reply and returns early

All 393 unit tests pass. Satisfies requirements 1.3 and 1.4.
2026-05-28 05:50:42 +00:00

613 lines
21 KiB
C
Raw Blame History

#define _GNU_SOURCE
#include "ha.h"
#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.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/socket.h>
#include <time.h>
#include <unistd.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;
}
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*) */
}
/* ── 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: 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 */
/* Hold lock to make the role-check + demote atomic, preventing a
* concurrent promotion from racing between the two operations. */
pthread_mutex_lock(&ha->lock);
/* Re-check term under lock in case a concurrent promotion raced us */
my_term = atomic_load_explicit(&ha->term, memory_order_acquire);
if (my_term == 0 && sender_term == 0) {
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) at term=0",
sender_priority);
ha_do_demote(ha);
}
}
pthread_mutex_unlock(&ha->lock);
}
/*
* 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
/* ── Watchdog thread implementation ────────────────────────── */
void ha_check_heartbeat_nocluster(ha_state_t *ha)
{
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
/* Only standbys watch for primary heartbeat timeout */
if (role != CTRL_STANDBY) return;
/* If we have no primary endpoint, nothing to monitor */
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) {
/* Primary appears dead — set failover_requested; watchdog acts on it */
atomic_store_explicit(&ha->failover_requested, 1, memory_order_release);
}
}
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;
ha_check_heartbeat_nocluster(ha);
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:
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;
}
View git blame Copy permalink