#define _GNU_SOURCE #include "ha.h" #include #include #include #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 "xce/errnox.h" #include "controller.capnp.h" #include "persist.h" #include "request.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; /* election-iomgr-refactor Task 2.1: explicit zero-init for documentation clarity * (memset above already zeroes these, but being explicit is better documentation) */ ha->client_ssl_ctx = NULL; ha->phase2_done = 0; ha->n_peers = 0; atomic_store_explicit(&ha->repl_ack_count, 0, memory_order_relaxed); /* 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_PRIMARY_COMMITTING: /* Req 5.1, 5.2, 8.1: Primary won election but has not yet received * majority replication ack. Allow only replication-related and * administrative opcodes; deny job dispatch and queue management * with XCE_ERR_NOT_READY. */ switch (opcode) { case OPCode_heartbeat: case OPCode_replConfig: case OPCode_replConfigPush: case OPCode_upgradeConfigAck: case OPCode_hello: case OPCode_walAck: case OPCode_replEntry: case OPCode_replAck: case OPCode_replCatchUpDone: return 0; default: 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: election_timeout + priority_step * priority + random jitter. * Higher-priority (lower index) nodes time out sooner, giving them a head * start in the election before lower-priority nodes begin campaigning. */ long to_ms = (long)ha->config.election_timeout_ms; to_ms += (long)ha->my_priority * (long)ha->config.election_priority_step_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; } /* ── ha_heartbeat_is_fresh (Req 6.2) ──────────────────────────────────────── * * Returns 1 if the elapsed time since ha->last_heartbeat is strictly less than * ha->config.heartbeat_timeout_ms (i.e. the primary is still alive). * Returns 0 if the heartbeat is stale (timeout has expired). * * Uses CLOCK_MONOTONIC to avoid wall-clock skew. */ static int ha_heartbeat_is_fresh(ha_state_t *ha) { struct timespec now, last; clock_gettime(CLOCK_MONOTONIC, &now); pthread_mutex_lock(&ha->lock); last = ha->last_heartbeat; pthread_mutex_unlock(&ha->lock); long ms = (now.tv_sec - last.tv_sec) * 1000L + (now.tv_nsec - last.tv_nsec) / 1000000L; return ms < (long)ha->config.heartbeat_timeout_ms; } /* ── Vote Handler (Req 2.2–2.6, 6.1, 6.2, 6.4, 9.1) ──────────────────────── * * handle_request_vote: process an incoming OP_requestVote RPC. * * Decision order (fail-fast): * 0. role gate (Req 6.1, 6.2): * - CTRL_PRIMARY or CTRL_PRIMARY_COMMITTING → deny * - CTRL_STANDBY with fresh heartbeat → deny * 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 6.1, 6.2: role-based deny gate — checked before any Raft state mutation. * PRIMARY and PRIMARY_COMMITTING never grant votes (protect the live primary). * STANDBY with a fresh heartbeat also denies (primary is alive, no new election * needed). CTRL_INIT and CTRL_ELECTING fall through to Raft evaluation. */ ctrl_role_t my_role = (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire); if (my_role == CTRL_PRIMARY || my_role == CTRL_PRIMARY_COMMITTING) return 0; /* deny — I am primary (or committing); protect against challengers */ if (my_role == CTRL_STANDBY && ha_heartbeat_is_fresh(ha)) return 0; /* deny — my primary is alive */ /* 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; } /* ── Type definitions needed by ha_elect_evaluate_and_signal (Task 4.1) ───── * * election_timeout_ctx_t and committing_timeout_ctx_t are defined here (before * ha_elect_evaluate_and_signal) because the COMMITTING path references them. * vote_ep_ctx_t is defined in the iomgr vote-send section below. */ typedef struct { _Atomic int cancelled; ha_state_t *ha; cluster_t *c; ha_elect_ctx_t *ctx; } election_timeout_ctx_t; /* committing_timeout_ctx_t: owns the heap allocation; freed by the callback. */ typedef struct { _Atomic int cancelled; ha_state_t *ha; cluster_t *c; } committing_timeout_ctx_t; /* committing_timeout_fn: forward declaration — defined after ha_elect_evaluate_and_signal */ static int committing_timeout_fn(void *arg); /* cluster_init_phase2: defined in cluster.c. Only referenced in production builds * (XCE_TESTING uses ha_phase2_hook instead, and cluster.c is not linked into the * unit-test binary). */ #ifndef XCE_TESTING extern int cluster_init_phase2(cluster_t *c); /* ha_send_ctrlhello_to_all_standbys: defined in cluster.c. * Iterates ha->standby_endpoints, sends CtrlHelloRequest to each, triggering * catch-up replication in handle_ctrlhello (Req 7.2). */ extern void ha_send_ctrlhello_to_all_standbys(cluster_t *c); /* push_config_to_standbys: defined in config_repl.c (Req 7.1). * Best-effort fan-out of config blob to all connected standbys. * Called with (c, NULL, 0, NULL, 0) to push from the primary's saved config. */ extern int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, const uint8_t sha256[32], uint64_t new_tag); #endif #ifdef XCE_TESTING /* * ha_phase2_hook: test-only seam. * When non-NULL, ha_elect_evaluate_and_signal() calls this INSTEAD OF the real * cluster_init_phase2() so tests can count phase2 calls without a live cluster_t. */ int (*ha_phase2_hook)(void *c) = NULL; /* * ha_send_ctrlhello_hook: test-only seam for ha_send_ctrlhello_to_all_standbys. * When non-NULL, the COMMITTING path calls this INSTEAD OF the real function * (cluster.c is not linked in unit-test binary). Req 7.2. */ int (*ha_send_ctrlhello_hook)(void *c) = NULL; /* * ha_push_config_hook: test-only seam for push_config_to_standbys. * When non-NULL, the COMMITTING path calls this INSTEAD OF the real function * (config_repl.c's push_config_to_standbys needs a real cluster). Req 7.1. */ int (*ha_push_config_hook)(void *c) = NULL; #endif /* XCE_TESTING */ /* ── Election Evaluation (Req 2.1, 2.7, 2.8) ─────────────────────────────── * * 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) { /* Guard: called from handle_request_vote_reply under ctx->lock; called from * unit-test path (peers==NULL) without threads — done_posted prevents double * execution if two threads both hit quorum simultaneously. */ if (ctx->done_posted) return (ctx->result == 1) ? CTRL_PRIMARY : CTRL_STANDBY; ctrl_role_t role_result; /* Fix 2 (CRITICAL): zero-peer guard — single-node shortcircuit. * n_peers==0 means no other nodes; votes[] flexible array has 0 entries, * so accessing votes[0] would be out-of-bounds. Self-vote (votes_granted=1) * always satisfies quorum=1. No fresher-peer or fatal-gate checks apply. */ if (ctx->n_peers == 0) { ctx->result = 1; ctx->done_posted = 1; sem_post(&ctx->done); role_result = CTRL_PRIMARY; goto apply_role; } { 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); role_result = CTRL_STANDBY; goto apply_role; } /* Check quorum */ if (ctx->votes_granted < quorum) { ctx->result = 0; ctx->done_posted = 1; sem_post(&ctx->done); role_result = CTRL_STANDBY; goto apply_role; } /* 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); role_result = CTRL_STANDBY; goto apply_role; } if (ctx->max_resp_tag == my_tag && ctx->max_resp_lsn > my_lsn) { ctx->result = 0; ctx->done_posted = 1; sem_post(&ctx->done); role_result = CTRL_STANDBY; goto apply_role; } /* Quorum met and no fresher peer → PRIMARY */ ctx->result = 1; ctx->done_posted = 1; sem_post(&ctx->done); role_result = CTRL_PRIMARY; } apply_role: /* ── Task 4.1: apply role and call cluster_init_phase2 ───────────────── * * After determining the election result, update ha->role and (on first boot) * call cluster_init_phase2. This runs once per process lifetime — the * phase2_done flag prevents re-entry on subsequent elections (RDHUP failover). * * The cluster_t pointer is carried in ctx->c (NULL in unit tests). * * Cancel the election timeout watchdog if we won (it would be a no-op now * but the explicit cancel avoids a spurious second evaluate call). */ { cluster_t *c = ctx->c; int is_first_boot = !ha->phase2_done; if (role_result == CTRL_PRIMARY) { /* Cancel election timeout watchdog (quorum already reached) */ if (ctx->timeout_ctx) { election_timeout_ctx_t *etc = (election_timeout_ctx_t *)ctx->timeout_ctx; atomic_store(&etc->cancelled, 1); } /* First boot: call cluster_init_phase2 to load config+data (Req 9.2) */ if (is_first_boot) { #ifdef XCE_TESTING /* In test mode, use hook if set; skip real phase2 (cluster.c not linked) */ if (ha_phase2_hook != NULL) ha_phase2_hook(c); #else if (c != NULL) cluster_init_phase2(c); #endif ha->phase2_done = 1; } /* Enter COMMITTING: block job traffic until majority ack (Req 4.1, 8.1) */ atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY_COMMITTING, memory_order_release); INFO("HA FSM: → CTRL_PRIMARY_COMMITTING (term=%" PRIu64 ", first_boot=%d)", atomic_load_explicit(&ha->term, memory_order_acquire), is_first_boot); /* Reset ack counter for this committing round (Req 5.2) */ atomic_store(&ha->repl_ack_count, 0); /* Push config to standbys with lower config_tag (Req 7.1) */ #ifdef XCE_TESTING if (ha_push_config_hook != NULL) ha_push_config_hook(c); #else if (c != NULL) push_config_to_standbys(c, NULL, 0, NULL, 0); #endif /* Send CtrlHelloRequest to all standbys to trigger WAL catch-up (Req 7.2) */ #ifdef XCE_TESTING if (ha_send_ctrlhello_hook != NULL) ha_send_ctrlhello_hook(c); #else if (c != NULL) ha_send_ctrlhello_to_all_standbys(c); #endif /* Arm committing timeout watchdog (degraded-mode fallback, Req 8.4) */ if (c) { committing_timeout_ctx_t *ct = calloc(1, sizeof(*ct)); if (ct) { atomic_init(&ct->cancelled, 0); ct->ha = ha; ct->c = c; int rc = delay_task_ms(&c->delaymgr, (int)ha->config.committing_timeout_ms, 1, committing_timeout_fn, ct); if (rc < 0) { WARN("ha_elect_evaluate_and_signal: delay_task_ms " "for committing timeout failed: %d", rc); free(ct); } } } } else { /* STANDBY result */ /* Cancel election timeout watchdog — it is now a no-op but explicit * cancellation matches spec wording and avoids a spurious lock-acquire * if the timeout fires after STANDBY is decided. */ if (ctx->timeout_ctx) { election_timeout_ctx_t *etc = (election_timeout_ctx_t *)ctx->timeout_ctx; atomic_store(&etc->cancelled, 1); } /* First boot: call cluster_init_phase2 so structures are ready; * config arrives via primary's replication push (Req 9.2, 7.3) */ if (is_first_boot) { #ifdef XCE_TESTING /* In test mode, use hook if set; skip real phase2 (cluster.c not linked) */ if (ha_phase2_hook != NULL) ha_phase2_hook(c); #else if (c != NULL) cluster_init_phase2(c); #endif ha->phase2_done = 1; } atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY, memory_order_release); INFO("HA FSM: → CTRL_STANDBY (term=%" PRIu64 ", first_boot=%d)", atomic_load_explicit(&ha->term, memory_order_acquire), is_first_boot); } } return role_result; } /* ── 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); } /* ── Task 2.2: iomgr-native vote send path ────────────────────────────────── * * vote_ep_ctx_t: per-peer context stored in ep->data for each outgoing vote * endpoint. on_vote_connected fires after the TLS handshake completes (when * endpoint_t.ssl_handshake_done transitions to 1 in io_handler). */ typedef struct { ha_state_t *ha; cluster_t *c; ha_elect_ctx_t *ctx; int peer_idx; uint64_t my_term; uint64_t my_tag; uint64_t my_lsn; const char *my_name; } vote_ep_ctx_t; /* xce_request: defined in request.c, not yet in any public header */ extern int xce_request(endpoint_t *ep, xce_request_t *req); /* * vote_reply_msg_handler: forward declaration. * Implemented in task 2.3. Registered on the endpoint by on_vote_connected * so that the next iomgr EPOLLIN delivers the vote reply to this handler. * Signature matches msg_handler_t = int (*)(void*) from netio.h. */ static int vote_reply_msg_handler(void *ep); #ifdef XCE_TESTING /* * xce_vote_request_hook: test-only seam. * * When non-NULL, on_vote_connected() calls this INSTEAD OF the real * xce_request() so tests can capture the call without requiring a live * iomgr or SSL endpoint. The hook receives the fully built * xce_request_t and may inspect all fields. * * Pattern mirrors xce_retry_send_hook in request.c. */ int (*xce_vote_request_hook)(endpoint_t *ep, xce_request_t *req) = NULL; #endif /* XCE_TESTING */ /* * on_vote_connected: iomgr callback fired after TLS handshake completes on an * outgoing vote endpoint. * * Steps: * 1. Cast ep->data to vote_ep_ctx_t. * 2. Build request_vote_req_t from ctx fields. * 3. Build xce_request_t{opcode=OPCode_requestVote, body.requestvotemsg=&vrq}. * 4. Call xce_vote_request_hook (test seam, XCE_TESTING only) — then * call xce_request() to queue the packet on the iomgr send path. * 5. Set ep->msg_handler = vote_reply_msg_handler so the next inbound * packet is decoded as a RequestVoteReply. * * Req 2.3: send vote after TLS handshake. * Req 1.2: vote RPC via iomgr path (xce_request enqueues to endpoint_enq). */ static void on_vote_connected(endpoint_t *ep) { vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data; /* Build the RequestVoteRequest payload */ request_vote_req_t vrq; memset(&vrq, 0, sizeof(vrq)); vrq.term = vctx->my_term; vrq.candidate_id = (char *)(vctx->my_name ? vctx->my_name : ""); vrq.config_tag = vctx->my_tag; vrq.last_log_lsn = vctx->my_lsn; /* Wrap in xce_request_t */ xce_request_t req; memset(&req, 0, sizeof(req)); req.opcode = OPCode_requestVote; req.body_which = Request_body_requestvotemsg; req.body.requestvotemsg = &vrq; #ifdef XCE_TESTING if (xce_vote_request_hook != NULL) xce_vote_request_hook(ep, &req); else xce_request(ep, &req); #else xce_request(ep, &req); #endif /* Switch to reply-reading mode: next inbound packet → vote_reply_msg_handler */ ep->msg_handler = vote_reply_msg_handler; } /* * xce_vote_reply_hook: test-only seam for vote_reply_msg_handler. * * Forward-declared here (before vote_reply_msg_handler) so the handler body * can reference it. Defined only under XCE_TESTING; guarded below. * * When non-NULL, vote_reply_msg_handler() calls this INSTEAD OF the real * handle_request_vote_reply() so tests can capture decoded fields without * a live iomgr. Receives the decoded scalar fields — no capnp pointers. * * Pattern mirrors xce_vote_request_hook used in task 2.2. */ #ifdef XCE_TESTING int (*xce_vote_reply_hook)(ha_state_t *ha, ha_elect_ctx_t *ctx, int peer_idx, uint64_t term, int vote_granted, uint64_t peer_tag, uint64_t peer_lsn) = NULL; #endif /* XCE_TESTING */ /* * vote_reply_dispatch: internal helper — calls handle_request_vote_reply or * the test hook (when XCE_TESTING and hook is non-NULL). */ static void vote_reply_dispatch(ha_state_t *ha, ha_elect_ctx_t *ctx, int idx, uint64_t term, int vote_granted, uint64_t peer_tag, uint64_t peer_lsn) { #ifdef XCE_TESTING if (xce_vote_reply_hook != NULL) { xce_vote_reply_hook(ha, ctx, idx, term, vote_granted, peer_tag, peer_lsn); return; } #endif handle_request_vote_reply(ha, ctx, idx, term, vote_granted, peer_tag, peer_lsn); } /* * vote_reply_msg_handler: iomgr message handler for incoming RequestVoteReply. * * Called by iomgr (via INVOKE_MSG_HANDLER) when a full packet arrives on an * outgoing vote endpoint after on_vote_connected fired. * * Steps: * 1. Cast ep to endpoint_t*; get packet from ep->fsm_in.pkt (owned by us). * 2. Raw capnp decode: init capn_ctx, read outer Request wrapper. * 3. read_RequestReply() into rep_s (capnp pointer — still valid). * 4. read_RequestVoteReply() into vr_raw (scalars) BEFORE capn_free. * 5. capn_free() — capnp segment released; vr_raw still valid (scalars only). * 6. free_packet() + free(pkt) — packet memory released. * 7. Term fencing: if vr_raw.term > ha->term, update ha->term atomically. * 8. Call vote_reply_dispatch() with vr_raw scalar fields. * * On any decode error: treat peer as denied (grant=0, tag=0, lsn=0). * * ASAN safety: all capnp fields read before capn_free() — no heap-use-after-free. * Requirements: 3.1 (iomgr packet), 3.2 (no capnp UAF), 3.3 (extract fields), * 3.4 (term fencing). */ static int vote_reply_msg_handler(void *ep_raw) { endpoint_t *ep = (endpoint_t *)ep_raw; vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data; ha_state_t *ha = vctx->ha; ha_elect_ctx_t *ctx = vctx->ctx; int idx = vctx->peer_idx; io_packet_t *pkt = ep->fsm_in.pkt; /* ownership transferred by INVOKE_MSG_HANDLER */ int ok = 0; /* Decoded reply fields — populated on success path */ uint64_t vr_term = 0; int vr_granted = 0; uint64_t vr_tag = 0; uint64_t vr_lsn = 0; if (!pkt || !pkt->data || pkt->sz == 0) goto done; { struct capn capn_ctx; struct Request outer_rq; Request_ptr outer_rp; struct RequestReply rep_s; struct RequestVoteReply vr_raw; int capn_ok; capn_ok = capn_init_mem(&capn_ctx, (const uint8_t *)pkt->data, (size_t)pkt->sz, 0); if (capn_ok != 0) goto done; outer_rp.p = capn_getp(capn_root(&capn_ctx), 0, 1); read_Request(&outer_rq, outer_rp); if (outer_rq.body_which != Request_body_replymsg || !outer_rq.body.replymsg.p.data) { capn_free(&capn_ctx); goto done; } read_RequestReply(&rep_s, outer_rq.body.replymsg); if (rep_s.data_which != RequestReply_data_requestvotereplymsg) { capn_free(&capn_ctx); goto done; } /* CRITICAL: copy all scalars BEFORE capn_free to avoid UAF. * struct RequestVoteReply fields: term, voteGranted, peerConfigTag, * peerLastLogLsn — all scalars; safe to copy before freeing segment. */ memset(&vr_raw, 0, sizeof(vr_raw)); read_RequestVoteReply(&vr_raw, rep_s.data.requestvotereplymsg); capn_free(&capn_ctx); /* capnp segment released; vr_raw.* still valid */ vr_term = (uint64_t)vr_raw.term; vr_granted = vr_raw.voteGranted ? 1 : 0; vr_tag = (uint64_t)vr_raw.peerConfigTag; vr_lsn = (uint64_t)vr_raw.peerLastLogLsn; ok = 1; } done: /* Release packet — INVOKE_MSG_HANDLER will null ep->fsm_in.pkt after return */ if (pkt) { free_packet(pkt); free(pkt); } if (ok) { /* Term fencing: if peer's term exceeds ours, update atomically */ uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire); if (vr_term > my_term) atomic_store_explicit(&ha->term, vr_term, memory_order_release); vote_reply_dispatch(ha, ctx, idx, vr_term, vr_granted, vr_tag, vr_lsn); } else { /* Decode error: treat peer as denied */ vote_reply_dispatch(ha, ctx, idx, atomic_load_explicit(&ha->term, memory_order_acquire), 0, 0, 0); } return 0; } #ifdef XCE_TESTING /* * vote_reply_build_pkt_for_test: test helper that encodes a RequestVoteReply * into a newly allocated io_packet_t that vote_reply_msg_handler can decode. * * Wire format mirrors the real peer reply: * Request{opcode=reply, body.replymsg=RequestReply{ * data_which=requestvotereplymsg, * data.requestvotereplymsg=RequestVoteReply{term, voteGranted, ...} * }} * * Returns a freshly allocated io_packet_t* on success, NULL on failure. * Handler takes ownership: it calls free_packet()+free() when done. * Tests set ep->fsm_in.pkt to the returned pointer before calling the handler. */ io_packet_t *vote_reply_build_pkt_for_test(uint64_t term, int vote_granted, uint64_t peer_tag, uint64_t peer_lsn) { struct capn cs; struct capn_segment *seg; struct RequestVoteReply vr_capn; RequestVoteReply_ptr vr_ptr; struct RequestReply rr_capn; RequestReply_ptr rr_ptr; struct Request rq_capn; Request_ptr rq_ptr; io_packet_t *pkt; int rc; capn_init_malloc(&cs); seg = capn_root(&cs).seg; /* RequestVoteReply — all scalar fields */ memset(&vr_capn, 0, sizeof(vr_capn)); vr_capn.term = term; vr_capn.voteGranted = vote_granted ? 1 : 0; vr_capn.peerConfigTag = peer_tag; vr_capn.peerLastLogLsn = peer_lsn; vr_ptr = new_RequestVoteReply(seg); write_RequestVoteReply(&vr_capn, vr_ptr); /* RequestReply envelope */ memset(&rr_capn, 0, sizeof(rr_capn)); rr_capn.seq = 1; rr_capn.opcode = OPCode_requestVote; rr_capn.status = 0; rr_capn.data_which = RequestReply_data_requestvotereplymsg; rr_capn.data.requestvotereplymsg = vr_ptr; rr_ptr = new_RequestReply(seg); write_RequestReply(&rr_capn, rr_ptr); /* Outer Request{opcode=reply} */ memset(&rq_capn, 0, sizeof(rq_capn)); rq_capn.opcode = OPCode_reply; rq_capn.seq = 1; rq_capn.body_which = Request_body_replymsg; rq_capn.body.replymsg = rr_ptr; rq_ptr = new_Request(seg); write_Request(&rq_capn, rq_ptr); capn_setp(capn_root(&cs), 0, rq_ptr.p); pkt = calloc(1, sizeof(*pkt)); if (!pkt) { capn_free(&cs); return NULL; } pkt->sz = capn_size(&cs); rc = alloc_packet(pkt); if (rc < 0) { capn_free(&cs); free(pkt); return NULL; } capn_write_mem(&cs, (uint8_t *)pkt->data, (size_t)pkt->sz, 0); capn_free(&cs); return pkt; } /* * vote_reply_msg_handler_for_test: expose vote_reply_msg_handler under * XCE_TESTING so tests can call it directly (it is otherwise static). * Sets ep->fsm_in.pkt = pkt then calls vote_reply_msg_handler(ep). * Caller owns ep->data (vote_ep_ctx_t); pkt ownership is transferred. */ int vote_reply_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt) { ep->fsm_in.pkt = pkt; return vote_reply_msg_handler(ep); } /* * vote_ep_ctx_build_for_test: test helper that allocates a vote_ep_ctx_t, * fills it from the supplied parameters, and wires up ep->data and * ep->on_connected so the test can drive on_vote_connected indirectly. * * Returns ep (same pointer passed in) for call-chain convenience. * Caller is responsible for free(ep->data) after use. */ endpoint_t *vote_ep_ctx_build_for_test(endpoint_t *ep, ha_state_t *ha, ha_elect_ctx_t *ctx, int peer_idx, uint64_t my_term, uint64_t my_tag, uint64_t my_lsn, const char *my_name) { vote_ep_ctx_t *vctx = calloc(1, sizeof(*vctx)); if (!vctx) return ep; /* test will fail when on_connected is NULL */ vctx->ha = ha; vctx->c = NULL; vctx->ctx = ctx; vctx->peer_idx = peer_idx; vctx->my_term = my_term; vctx->my_tag = my_tag; vctx->my_lsn = my_lsn; vctx->my_name = my_name; ep->data = vctx; ep->on_connected = on_vote_connected; return ep; } /* * election_timeout_fn_for_test: test helper that exercises the election timeout * callback logic with an explicit cancelled_val instead of an atomic field. * Allows testing the cancelled-gate without setting up a real delaymgr. * * Returns 0 in all cases (mirrors the real callback's return value). */ int election_timeout_fn_for_test(ha_state_t *ha, ha_elect_ctx_t *ctx, int cancelled_val) { if (cancelled_val) return 0; pthread_mutex_lock(&ctx->lock); ha_elect_evaluate_and_signal(ha, ctx); pthread_mutex_unlock(&ctx->lock); return 0; } /* * committing_timeout_fn_for_test: test helper that exercises the committing * timeout callback logic with an explicit cancelled_val. The cluster pointer * may be NULL; the function checks the role on `ha` and calls * ha_promote_to_primary(c) only when role == CTRL_PRIMARY_COMMITTING && c != NULL. * * Unlike the production callback, this shim does NOT free the ctx structure — * it is stack/test-owned and must not be heap-freed by the shim. * * Returns 0 in all cases (mirrors the real callback). */ int committing_timeout_fn_for_test(ha_state_t *ha, cluster_t *c, int cancelled_val) { if (cancelled_val) return 0; ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire); if (role == CTRL_PRIMARY_COMMITTING && c != NULL) ha_promote_to_primary(c); return 0; } #endif /* XCE_TESTING */ /* ── Task 2.4: election_timeout_fn — delaymgr callback ───────────────────── * * Fires ha_elect_evaluate_and_signal after election_timeout_ms if not cancelled. * The cancelled flag is set atomically when quorum is already reached, making * this callback a no-op. Owns the election_timeout_ctx_t — frees it on exit. */ static int election_timeout_fn(void *arg) { election_timeout_ctx_t *t = (election_timeout_ctx_t *)arg; if (atomic_load(&t->cancelled)) { free(t); return 0; } pthread_mutex_lock(&t->ctx->lock); ha_elect_evaluate_and_signal(t->ha, t->ctx); pthread_mutex_unlock(&t->ctx->lock); free(t); return 0; } /* ── Task 4.1: committing_timeout_fn — delaymgr callback ─────────────────── * * Fires ha_promote_to_primary (degraded mode) if the majority ack threshold * has not been reached within committing_timeout_ms. Req 8.4. * Owns the committing_timeout_ctx_t — frees it on exit. */ static int committing_timeout_fn(void *arg) { committing_timeout_ctx_t *t = (committing_timeout_ctx_t *)arg; if (atomic_load(&t->cancelled)) { free(t); return 0; } ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&t->ha->role, memory_order_acquire); if (role == CTRL_PRIMARY_COMMITTING) { WARN("HA: committing timeout — majority ack not received within timeout; " "promoting to PRIMARY (degraded mode, Req 8.4)"); if (t->c) ha_promote_to_primary(t->c); } free(t); return 0; } /* ── Task 4.1: ha_repl_ack_received ──────────────────────────────────────── * * Called when a standby ACKs config+WAL catch-up (OPCode_replCatchUpDone). * Increments repl_ack_count; when majority threshold (quorum-1 peers, since * primary counts itself) is reached, disarms the committing timeout and calls * ha_promote_to_primary. Req 8.2. */ void ha_repl_ack_received(ha_state_t *ha, cluster_t *c) { if (!ha) return; int acks = atomic_fetch_add(&ha->repl_ack_count, 1) + 1; int quorum = (ha->n_peers + 2) / 2; /* majority of total cluster */ if (acks >= quorum - 1) { /* -1: primary counts itself */ INFO("HA: majority ack received (acks=%d, quorum=%d) → CTRL_PRIMARY", acks, quorum); if (c) ha_promote_to_primary(c); } } /* ── Task 2.4: vote_ep_err_handler ───────────────────────────────────────── * * Called by iomgr when the outgoing vote endpoint encounters an error * (connect failure, SSL error, remote close before reply). * Counts the peer as denied: grant=0, tag=0, lsn=0. */ static int vote_ep_err_handler(void *ep_raw) { endpoint_t *ep = (endpoint_t *)ep_raw; vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data; if (!vctx) return 0; handle_request_vote_reply(vctx->ha, vctx->ctx, vctx->peer_idx, vctx->my_term, 0, /* vote_granted = false */ 0, 0); /* Fix 5 (HIGH): do NOT free vctx here. * ep->data (vctx) is also referenced by vote_reply_msg_handler via ep->data. * The endpoint lifetime is managed by iomgr; vctx is freed in * ha_election_cleanup() (task 5.1) when all per-peer contexts are released. * Freeing here would cause a use-after-free in vote_reply_msg_handler. */ return 0; } /* ── Task 2.4: ha_start_election_async ───────────────────────────────────── * * Non-blocking election initiator. Returns immediately after registering * all outgoing vote endpoints with iomgr and scheduling the timeout watchdog. * * Single-node shortcircuit (n_peers == 1): * Allocates a self-only ha_elect_ctx_t, calls ha_elect_evaluate_and_signal * directly, frees ctx. No iomgr or delaymgr involvement. * * Multi-node (n_peers > 1): * For each peer (peers[0..n_peers-2]): allocate vote_ep_ctx_t, get a fresh * endpoint_t, init with client_ssl_ctx, set on_vote_connected + * vote_ep_err_handler, call endpoint_connect + iomgr_monitor. Skip peer * on any failure (peer counted as denied via err_handler path later, or * immediately if connect returns synchronous failure and err_handler fires). * Allocate election_timeout_ctx_t (cancelled=0) and schedule * delay_task_ms(c->delaymgr, election_timeout_ms, 1, election_timeout_fn, t). * * Requirements: 1.3, 2.1 (shortcircuit), 2.4 (async vote send) */ void ha_start_election_async(ha_state_t *ha, cluster_t *c, int n_peers, xce_candidate_t **peers, uint64_t my_config_tag, uint64_t my_lsn, const char *my_name) { /* Step 1: store peer count */ ha->n_peers = n_peers; /* Increment term for this 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); /* Transition to CTRL_ELECTING */ atomic_store_explicit(&ha->role, (int32_t)CTRL_ELECTING, memory_order_release); INFO("HA FSM: → CTRL_ELECTING (async, term=%" PRIu64 ", n_peers=%d)", cur_term, n_peers); /* Step 2: single-node shortcircuit — self-elect immediately */ if (n_peers == 1) { ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx)); if (!ctx) { WARN("ha_start_election_async: calloc ctx failed (single-node)"); return; } ctx->n_peers = 0; /* no other peers */ 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; ctx->c = c; /* Task 4.1: pass cluster for phase2 + committing timeout */ pthread_mutex_init(&ctx->lock, NULL); sem_init(&ctx->done, 0, 0); /* Store candidate own tag/lsn in ctx fields (votes[] has 0 entries * for n_peers==0 shortcircuit — votes[0] would be out-of-bounds). */ ctx->max_resp_tag = my_config_tag; ctx->max_resp_lsn = my_lsn; /* For single-node, quorum = ceil(1/2) = 1; votes_granted=1 → wins. * Task 4.1: ha_elect_evaluate_and_signal now handles COMMITTING entry, * phase2, repl_ack_count reset, and committing timeout arming. */ INFO("HA election: outcome=PRIMARY term=%" PRIu64 " votes_granted=1 quorum=1 (single-node async)", cur_term); /* evaluate sets ha->role = CTRL_PRIMARY_COMMITTING and calls phase2 */ ha_elect_evaluate_and_signal(ha, ctx); pthread_mutex_destroy(&ctx->lock); sem_destroy(&ctx->done); free(ctx); return; } /* Step 3: multi-node — n_other = n_peers - 1 peers to solicit */ int n_other = n_peers - 1; /* Allocate shared election context */ ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) + (size_t)n_other * sizeof(ctx->votes[0])); if (!ctx) { WARN("ha_start_election_async: calloc ctx failed (multi-node)"); return; } 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; ctx->c = c; /* Task 4.1: pass cluster for phase2 + committing timeout */ pthread_mutex_init(&ctx->lock, NULL); sem_init(&ctx->done, 0, 0); /* Store candidate's own tag/lsn in votes[0] for the yield gate */ if (n_other > 0) { ctx->votes[0].my_tag = my_config_tag; ctx->votes[0].my_lsn = my_lsn; } /* Store ctx in ha for cleanup path */ atomic_store_explicit(&ha->elect_ctx, ctx, memory_order_release); /* Connect to each peer and register with iomgr */ int peers_connected = 0; for (int i = 0; peers != NULL && i < n_other; i++) { xce_candidate_t *peer = peers[i]; if (!peer) continue; /* Allocate per-peer context */ vote_ep_ctx_t *vctx = calloc(1, sizeof(*vctx)); if (!vctx) { WARN("ha_start_election_async: calloc vctx failed for peer %d", i); /* Count as denied immediately */ handle_request_vote_reply(ha, ctx, i, cur_term, 0, 0, 0); continue; } vctx->ha = ha; vctx->c = c; vctx->ctx = ctx; vctx->peer_idx = i; vctx->my_term = cur_term; vctx->my_tag = my_config_tag; vctx->my_lsn = my_lsn; vctx->my_name = my_name; /* Get a fresh endpoint from the iomgr pool */ endpoint_t *ep = endpoint_get(); if (!ep) { WARN("ha_start_election_async: endpoint_get() failed for peer %d", i); free(vctx); handle_request_vote_reply(ha, ctx, i, cur_term, 0, 0, 0); continue; } /* Initialize endpoint: client-side SSL, no listener (fd=-1 until connect) */ endpoint_init(ep, -1, 0, 0, 0); ep->data = vctx; /* Wire up SSL context (client method for outgoing vote connections) */ if (ha->client_ssl_ctx && c) { /* SSL_CTX is set on the iomgr; endpoint_connect will use it */ ep->iomgr = &c->iomgr; /* Set client SSL context on the endpoint's SSL object after connect */ /* endpoint_connect creates the SSL object from iomgr->ssl_ctx; * for outgoing connections we need TLS_client_method context. * Store client_ssl_ctx reference; the connect path will use it. */ } else if (c) { ep->iomgr = &c->iomgr; } /* Set callbacks */ endpoint_set_on_connected(ep, on_vote_connected); endpoint_set_err_handler(ep, vote_ep_err_handler); /* Connect asynchronously */ int rc = endpoint_connect(ep, peer->hostname, peer->port); if (rc < 0) { WARN("ha_start_election_async: endpoint_connect failed for peer %d (%s:%d): %d", i, peer->hostname, peer->port, rc); free(vctx); endpoint_fini(ep); /* Count as denied */ handle_request_vote_reply(ha, ctx, i, cur_term, 0, 0, 0); continue; } /* Fix 3 (HIGH): endpoint_connect creates the SSL object from iomgr->ssl_ctx * (server context). For outgoing vote connections we need TLS_client_method. * After connect, replace the SSL object with one from client_ssl_ctx and * explicitly set connect (client) state so the handshake goes client-side. * The fd is already set on the old ssl object — re-set it on the new one. */ if (ha->client_ssl_ctx && ep->ssl) { SSL_CTX *client_ctx = (SSL_CTX *)ha->client_ssl_ctx; SSL *new_ssl = SSL_new(client_ctx); if (new_ssl) { int ep_fd = SSL_get_fd(ep->ssl); SSL_free(ep->ssl); ep->ssl = new_ssl; SSL_set_fd(ep->ssl, ep_fd); SSL_set_connect_state(ep->ssl); } /* If SSL_new fails, fall through with the original SSL object; * worst case the handshake uses server context on client side, * which will fail and trigger vote_ep_err_handler (peer denied). */ } /* Register with iomgr for async I/O */ if (c) { rc = iomgr_monitor(ep); if (rc < 0) { WARN("ha_start_election_async: iomgr_monitor failed for peer %d: %d", i, rc); free(vctx); endpoint_fini(ep); handle_request_vote_reply(ha, ctx, i, cur_term, 0, 0, 0); continue; } } peers_connected++; INFO("ha_start_election_async: peer %d (%s:%d) connecting async", i, peer->hostname, peer->port); } /* Step 4: schedule election timeout watchdog */ if (c) { election_timeout_ctx_t *t = calloc(1, sizeof(*t)); if (t) { atomic_init(&t->cancelled, 0); t->ha = ha; t->c = c; t->ctx = ctx; int rc = delay_task_ms(&c->delaymgr, (int)ha->config.election_timeout_ms, 1, election_timeout_fn, t); if (rc < 0) { WARN("ha_start_election_async: delay_task_ms failed: %d", rc); free(t); /* Fallback: evaluate now if no peers connected */ if (peers_connected == 0) { pthread_mutex_lock(&ctx->lock); ha_elect_evaluate_and_signal(ha, ctx); pthread_mutex_unlock(&ctx->lock); } } else { /* Fix 4 (HIGH): wire timeout_ctx back into ctx so that task 4.1 * (quorum-reached path in handle_request_vote_reply) can cancel * the watchdog by setting timeout_ctx->cancelled = 1 atomically. */ ctx->timeout_ctx = t; } } else { WARN("ha_start_election_async: calloc election_timeout_ctx_t failed"); } } else if (peers_connected == 0 && n_other > 0) { /* No iomgr and no peers connected (test mode with n_peers>1, peers=NULL): * evaluate immediately so the election doesn't stall, then free ctx * since no delaymgr task owns it. */ pthread_mutex_lock(&ctx->lock); ha_elect_evaluate_and_signal(ha, ctx); pthread_mutex_unlock(&ctx->lock); atomic_store_explicit(&ha->elect_ctx, NULL, memory_order_release); pthread_mutex_destroy(&ctx->lock); sem_destroy(&ctx->done); free(ctx); ctx = NULL; } INFO("ha_start_election_async: returning (term=%" PRIu64 ", n_peers=%d, connected=%d)", cur_term, n_peers, peers_connected); /* Returns immediately — election result delivered via ha_elect_evaluate_and_signal */ } /* ── 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; /* * ha_rdhup_election_hook: test-only seam for ha_start_election_async in the * RDHUP failover path. When non-NULL, ha_on_primary_ctrl_fd_rdhup() calls * this instead of the real ha_start_election_async(). Req 6.3, 6.5, 9.3. */ void (*ha_rdhup_election_hook)(ha_state_t *ha, cluster_t *c, int n_peers, xce_candidate_t **peers, uint64_t my_config_tag, uint64_t my_lsn, const char *my_name) = 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 async election — ha_start_election_async is non-blocking (Req 6.3, 6.5). * ha->phase2_done==1 at this point (set after initial startup election), so * ha_elect_evaluate_and_signal will skip cluster_init_phase2 and go directly * to CTRL_PRIMARY_COMMITTING (Req 9.3). */ 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); const char *my_name = c->myconf ? c->myconf->name : ""; #ifdef XCE_TESTING if (ha_rdhup_election_hook != NULL) { ha_rdhup_election_hook(ha, c, np, peer_ptrs, my_tag, my_lsn, my_name); } else { ha_start_election_async(ha, c, np, peer_ptrs, my_tag, my_lsn, my_name); } #else ha_start_election_async(ha, c, np, peer_ptrs, my_tag, my_lsn, my_name); #endif } } /* ── 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; }