myxce/src/controller/ha.c
Rongsong Shen aab41fa609 feat(ha-repl-hardening): per-standby convergence tracking from heartbeats (task 3.1)
ha_process_standby_state stores (tag, lsn) per standby, evaluates
convergence (tag>=committed && lsn>=committed), sets converged flag,
logs INFO only on transition. Wired into repl_ack_decode_and_dispatch.
5 tests: converged, lsn-behind, tag-behind, regression, multi-peer.
2026-06-04 16:40:44 +00:00

3226 lines
123 KiB
C
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#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 "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;
/* No-op stubs removed: s_real_cbs (below) is used exclusively since task 6.2. */
/* ── Forward declarations needed by real raft callbacks (Task 6.2) ────────
* These static functions and types are defined later in this file; declared
* here so the callback implementations can reference them. */
/* vote_ep_ctx_t: per-peer context for outgoing vote endpoints (full def below) */
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;
/* election_timeout_ctx_t: delaymgr callback context for raft election timeout */
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;
/* cluster_init_phase2 / push_config_to_standbys / ha_send_ctrlhello_to_all_standbys:
* defined in cluster.c / config_repl.c; only referenced in production builds.
* (XCE_TESTING uses hook shims and does not link cluster.c.) */
#ifndef XCE_TESTING
extern int cluster_init_phase2(cluster_t *c);
extern void ha_send_ctrlhello_to_all_standbys(cluster_t *c);
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
static void on_vote_connected(endpoint_t *ep);
static int vote_ep_err_handler(void *ep_raw);
static int committing_timeout_fn(void *arg);
static int ha_heartbeat_is_fresh(ha_state_t *ha);
static int primary_heartbeat_fn(void *arg);
static void on_standby_connected(endpoint_t *ep);
/* ── Real raft callback implementations (Task 6.2) ─────────────
*
* These are registered as s_real_cbs in ha_init, replacing the no-op stubs.
* They are wired into the active election path in task 6.3.
*
* All callbacks receive ctx = ha_state_t*.
* ha->raft_c is set by ha_start_election_async before any raft_* calls.
* When ha->raft_c is NULL (unit test mode), callbacks that require cluster_t
* are no-ops.
*/
/*
* raft_cb_send_vote_request: encode and send a capnp VoteRequest to peer.
*
* peer_idx is the absolute candidate index in c->myconf->peers_cfg.candidates[].
* Uses the existing iomgr vote-send path (on_vote_connected / vote_ep_ctx_t).
* Reads the current election context from ha->elect_ctx (set before raft_start_election).
* Returns 0 on success; non-zero if no cluster context or alloc failure.
*/
static int raft_cb_send_vote_request(void *ctx, int peer_idx, uint64_t term)
{
ha_state_t *ha = (ha_state_t *)ctx;
cluster_t *c = ha->raft_c;
if (!c || !c->myconf) return 1; /* test mode */
const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
if (peer_idx < 0 || peer_idx >= pcfg->candidates_count)
return 1;
ha_elect_ctx_t *ectx = (ha_elect_ctx_t *)atomic_load_explicit(
&ha->elect_ctx, memory_order_acquire);
if (!ectx) return 1; /* no election in progress */
xce_candidate_t *peer = (xce_candidate_t *)&pcfg->candidates[peer_idx];
/* Compute the local index of peer_idx within ectx->votes[]:
* votes[] has n_peers - 1 = n_other entries for peers excluding self.
* Map: votes_idx = peer_idx if peer_idx < self_idx, else peer_idx - 1. */
int self_idx = ha->my_priority; /* self_idx == my_priority by convention */
int votes_idx = (peer_idx < self_idx) ? peer_idx : peer_idx - 1;
if (votes_idx < 0 || votes_idx >= ectx->n_peers)
return 1;
/* Allocate per-peer context */
vote_ep_ctx_t *vctx = calloc(1, sizeof(*vctx));
if (!vctx) {
handle_request_vote_reply(ha, ectx, votes_idx, term, 0, 0, 0);
return 1;
}
vctx->ha = ha;
vctx->c = c;
vctx->ctx = ectx;
vctx->peer_idx = votes_idx;
vctx->my_term = term;
/* tag and lsn come from votes[0] which holds our own values */
vctx->my_tag = (ectx->n_peers > 0) ? ectx->votes[0].my_tag : 0;
vctx->my_lsn = (ectx->n_peers > 0) ? ectx->votes[0].my_lsn : 0;
vctx->my_name = c->myconf ? c->myconf->name : "";
/* Get a fresh endpoint */
endpoint_t *ep = endpoint_get();
if (!ep) {
free(vctx);
handle_request_vote_reply(ha, ectx, votes_idx, term, 0, 0, 0);
return 1;
}
endpoint_init(ep, AF_INET, SOCK_STREAM, 0, 0);
if (c) endpoint_set_peer(ep, c->myhost);
ep->data = vctx;
ep->iomgr = &c->iomgr;
if (ha->client_ssl_ctx) {
SSL_CTX *client_ctx = (SSL_CTX *)ha->client_ssl_ctx;
ep->ssl = SSL_new(client_ctx);
if (ep->ssl)
SSL_set_connect_state(ep->ssl);
}
endpoint_set_on_connected(ep, on_vote_connected);
endpoint_set_err_handler(ep, vote_ep_err_handler);
int rc = endpoint_connect(ep, peer->hostname, peer->port);
if (rc < 0) {
endpoint_fini(ep);
if (!c)
handle_request_vote_reply(ha, ectx, votes_idx, term, 0, 0, 0);
return 1;
}
rc = iomgr_monitor(ep);
if (rc < 0) {
endpoint_fini(ep); /* frees ep->data (vctx) via FREEUP */
handle_request_vote_reply(ha, ectx, votes_idx, term, 0, 0, 0);
return 1;
}
/* Track endpoint for promotion/close after election result */
ectx->votes[votes_idx].ep = ep;
return 0;
}
/*
* raft_cb_vote_eligible: freshness gate delegated from raft library.
*
* from_idx is the absolute candidate index of the requester.
* Applies the same checks as handle_request_vote:
* - role gate: PRIMARY or COMMITTING → deny
* - fresh heartbeat on STANDBY → deny
* - in_catch_up → deny
* - freshness: candidate config_tag/lsn must be ≥ ours
* Returns 1 to grant, 0 to deny.
*/
static int raft_cb_vote_eligible(void *ctx, int from_idx, uint64_t term)
{
(void)from_idx; (void)term;
ha_state_t *ha = (ha_state_t *)ctx;
/* Role gate */
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
if (role == CTRL_PRIMARY || role == CTRL_PRIMARY_COMMITTING)
return 0;
/* Fresh heartbeat on STANDBY → primary is alive, deny */
if (role == CTRL_STANDBY && ha_heartbeat_is_fresh(ha))
return 0;
/* In catch-up → deny */
if (atomic_load_explicit(&ha->in_catch_up, memory_order_acquire))
return 0;
/* Freshness gate requires knowing the candidate's config_tag/lsn.
* The raft library does not pass these; they are carried in the capnp
* RequestVoteRequest fields. For the adapter, freshness is checked in
* handle_request_vote (called from the iomgr decode path in task 6.3).
* Here we return 1 (eligible) so the raft library proceeds; the full
* freshness gate is enforced by the caller before invoking raft_on_vote_request.
* This is safe because the caller (task 6.3 dispatcher) already applies
* the freshness check from handle_request_vote before delegating here. */
return 1;
}
/* raft_election_timeout_fn: delaymgr callback for raft election timeout.
* Fires raft_on_election_timeout after delay_ms if not cancelled.
* Owns the election_timeout_ctx_t — frees it on exit. */
static int raft_election_timeout_fn(void *arg)
{
election_timeout_ctx_t *t = (election_timeout_ctx_t *)arg;
ha_state_t *ha = t->ha;
/* Clear the stored pointer before checking cancelled so on_cancel_timeout
* is a safe no-op if it races with this callback. */
ha->raft_election_timeout_ctx = NULL;
if (atomic_load(&t->cancelled)) {
free(t);
return 0;
}
raft_on_election_timeout(&ha->raft);
free(t);
return 0;
}
/*
* raft_cb_schedule_timeout: schedule election timeout via delaymgr.
*
* Creates an election_timeout_ctx_t and calls delay_task_ms.
* The pointer is stored in ha->raft_election_timeout_ctx for cancellation.
*/
static void raft_cb_schedule_timeout(void *ctx, uint32_t delay_ms)
{
ha_state_t *ha = (ha_state_t *)ctx;
cluster_t *c = ha->raft_c;
if (!c) return; /* test mode */
election_timeout_ctx_t *t = calloc(1, sizeof(*t));
if (!t) return;
atomic_init(&t->cancelled, 0);
t->ha = ha;
t->c = c;
t->ctx = NULL; /* raft path does not use ha_elect_ctx_t */
int rc = delay_task_ms(&c->delaymgr, (int)delay_ms, 1,
raft_election_timeout_fn, t);
if (rc < 0) {
WARN("raft_cb_schedule_timeout: delay_task_ms failed: %d", rc);
free(t);
return;
}
ha->raft_election_timeout_ctx = t;
}
/*
* raft_cb_cancel_timeout: cancel the pending election timeout.
* Sets cancelled=1 on the stored election_timeout_ctx_t.
*/
static void raft_cb_cancel_timeout(void *ctx)
{
ha_state_t *ha = (ha_state_t *)ctx;
election_timeout_ctx_t *t =
(election_timeout_ctx_t *)ha->raft_election_timeout_ctx;
if (t)
atomic_store(&t->cancelled, 1);
ha->raft_election_timeout_ctx = NULL;
}
/* raft_rdhup_delay_fn: delaymgr callback for RDHUP-triggered delayed election.
* Owns its ctx (election_timeout_ctx_t) — frees it on exit.
* Calls raft_check_and_start_election to start election only if not already voted. */
static int raft_rdhup_delay_fn(void *arg)
{
election_timeout_ctx_t *t = (election_timeout_ctx_t *)arg;
ha_state_t *ha = t->ha;
if (!atomic_load(&t->cancelled))
raft_check_and_start_election(&ha->raft);
free(t);
return 0;
}
/*
* raft_cb_start_election_after_delay: schedule RDHUP retry timer.
* After delay_ms, calls raft_check_and_start_election to start election
* only if this node hasn't already voted for another candidate.
*/
static void raft_cb_start_election_after_delay(void *ctx, uint32_t delay_ms)
{
ha_state_t *ha = (ha_state_t *)ctx;
cluster_t *c = ha->raft_c;
if (!c) return; /* test mode */
/* Priority 0 → delay_ms == 0: highest-priority node starts immediately. */
if (delay_ms == 0) {
raft_check_and_start_election(&ha->raft);
return;
}
election_timeout_ctx_t *t = calloc(1, sizeof(*t));
if (!t) return;
atomic_init(&t->cancelled, 0);
t->ha = ha;
t->c = c;
t->ctx = NULL;
int rc = delay_task_ms(&c->delaymgr, (int)delay_ms, 1,
raft_rdhup_delay_fn, t);
if (rc < 0) {
WARN("raft_cb_start_election_after_delay: delay_task_ms failed: %d", rc);
free(t);
}
}
/*
* raft_cb_become_leader: node won the election — enter COMMITTING state.
*
* Equivalent of ha_elect_evaluate_and_signal CTRL_PRIMARY path:
* - enter CTRL_PRIMARY_COMMITTING
* - call cluster_init_phase2 on first boot
* - reset repl_ack_count, push config, send ctrlhello
* - arm committing timeout watchdog
*
* ha->raft_c must be non-NULL (production path only; test mode uses noop stubs).
*/
static void raft_cb_become_leader(void *ctx, uint64_t term)
{
ha_state_t *ha = (ha_state_t *)ctx;
cluster_t *c = ha->raft_c;
/* Cancel any pending election timeout */
raft_cb_cancel_timeout(ctx);
/* Sync ha->term from raft term (raft owns term increments) */
atomic_store_explicit(&ha->term, term, memory_order_release);
int is_first_boot = !ha->phase2_done;
/* First boot: call cluster_init_phase2 */
if (is_first_boot) {
#ifdef XCE_TESTING
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 */
atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
INFO("HA FSM (raft): → CTRL_PRIMARY_COMMITTING (term=%" PRIu64
", first_boot=%d)", term, is_first_boot);
/* Reset ack counter for this committing round */
atomic_store(&ha->repl_ack_count, 0);
/* Promote surviving vote endpoints to standby replication channels.
* Iterate ha->elect_ctx->votes[i].ep and add non-NULL entries to
* ha->standby_endpoints so push_config and ctrlhello reach them. */
{
ha_elect_ctx_t *ectx = (ha_elect_ctx_t *)atomic_load_explicit(
&ha->elect_ctx, memory_order_acquire);
if (ectx && ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
for (int i = 0; i < ectx->n_peers; i++) {
if (ectx->votes[i].ep != NULL) {
zarray_store(ha->standby_endpoints, ectx->votes[i].ep);
INFO("HA: promoted vote ep[%d] → standby channel", i);
}
}
pthread_mutex_unlock(&ha->lock);
}
}
if (c) {
/* Push config to standbys — read raw bytes from disk so standbys
* can save the file. NULL bytes are passed only in test mode. */
#ifdef XCE_TESTING
if (ha_push_config_hook != NULL)
ha_push_config_hook(c);
#else
{
uint8_t *cfg_bytes = NULL;
size_t cfg_len = 0;
extern int conf_read_raw(const char *, uint8_t **, size_t *);
if (c->myconf && c->myconf->conffile &&
conf_read_raw(c->myconf->conffile, &cfg_bytes, &cfg_len) == 0) {
push_config_to_standbys(c, cfg_bytes, cfg_len,
c->myconf->config_sha256,
c->myconf->config_version_tag);
free(cfg_bytes);
} else {
push_config_to_standbys(c, NULL, 0, NULL,
c->myconf ? c->myconf->config_version_tag : 0);
}
}
#endif
/* Send CtrlHelloRequest to all standbys to trigger WAL catch-up */
#ifdef XCE_TESTING
if (ha_send_ctrlhello_hook != NULL)
ha_send_ctrlhello_hook(c);
#else
ha_send_ctrlhello_to_all_standbys(c);
#endif
/* Start primary heartbeat immediately in COMMITTING so standbys receive
* NOP during the committing window and their liveness poll does not fire.
* ha_start_primary_heartbeat is also called in ha_promote_to_primary;
* the idempotency guard (role check at fn top) prevents double-arming. */
ha_start_primary_heartbeat(c);
/* Arm committing timeout watchdog */
{
/* Cancel any prior committing timeout (e.g. from a previous term) */
committing_timeout_ctx_t *prev = (committing_timeout_ctx_t *)
atomic_exchange(&ha->committing_timeout_ctx, NULL);
if (prev) {
atomic_store(&prev->cancelled, 1);
}
}
committing_timeout_ctx_t *ct = calloc(1, sizeof(*ct));
if (ct) {
atomic_init(&ct->cancelled, 0);
ct->ha = ha;
ct->c = c;
atomic_store(&ha->committing_timeout_ctx, ct);
int rc = delay_task_ms(&c->delaymgr,
(int)ha->config.committing_timeout_ms,
1, committing_timeout_fn, ct);
if (rc < 0) {
WARN("raft_cb_become_leader: delay_task_ms for committing timeout failed: %d",
rc);
atomic_store(&ha->committing_timeout_ctx, NULL);
free(ct);
}
}
}
}
/*
* raft_cb_become_follower: node lost election or saw higher term — enter STANDBY.
*
* Syncs ha->term from the raft-reported term, sets role to CTRL_STANDBY,
* calls cluster_init_phase2 on first boot (standbys also need it to
* receive config replication push from the new primary).
*/
static void raft_cb_become_follower(void *ctx, uint64_t term, int leader_idx)
{
(void)leader_idx;
ha_state_t *ha = (ha_state_t *)ctx;
cluster_t *c = ha->raft_c;
/* Cancel any pending election timeout */
raft_cb_cancel_timeout(ctx);
/* Cancel any pending committing timeout — we are no longer leader.
* Without this, a node that stepped down due to a higher-term vote
* request would still promote to PRIMARY when the old committing
* timeout fires → split-brain. */
{
committing_timeout_ctx_t *ct = (committing_timeout_ctx_t *)
atomic_exchange(&ha->committing_timeout_ctx, NULL);
if (ct) {
atomic_store(&ct->cancelled, 1);
}
}
/* Sync ha->term */
atomic_store_explicit(&ha->term, term, memory_order_release);
int is_first_boot = !ha->phase2_done;
/* First boot: call cluster_init_phase2 so structures are ready */
if (is_first_boot) {
#ifdef XCE_TESTING
if (ha_phase2_hook != NULL)
ha_phase2_hook(c);
#else
if (c != NULL)
cluster_init_phase2(c);
#endif
ha->phase2_done = 1;
}
/* Close all outgoing vote endpoints (loser path).
* The primary will push config and ctrlhello via promoted connections;
* the standby waits passively on its listener for the inbound connection. */
{
ha_elect_ctx_t *ectx = (ha_elect_ctx_t *)atomic_load_explicit(
&ha->elect_ctx, memory_order_acquire);
if (ectx) {
for (int i = 0; i < ectx->n_peers; i++) {
if (ectx->votes[i].ep != NULL) {
INFO("HA: closing vote ep[%d] (loser path)", i);
endpoint_fini(ectx->votes[i].ep);
ectx->votes[i].ep = NULL;
}
}
}
}
atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY,
memory_order_release);
INFO("HA FSM (raft): → CTRL_STANDBY (term=%" PRIu64 ", first_boot=%d)",
term, is_first_boot);
/* Arm heartbeat watchdog — fires ha_on_primary_ctrl_fd_rdhup if primary
* sends no ctrlhello within election_timeout_ms. */
ha_arm_heartbeat_timeout_vtask(ha, c);
}
/*
* raft_cb_persist: persist term + voted_for to disk before granting vote.
*
* Maps raft integer voted_for index to a candidate name using cluster config.
* Updates ha->term, ha->voted_for_term, ha->voted_for_name then calls
* ha_persist_term(ha, workdir).
*
* voted_for == -1 means "no vote"; voted_for >= 0 is the candidate index.
* Returns 0 on success, non-zero on error (causes raft library to deny vote).
*/
static int raft_cb_persist(void *ctx, uint64_t term, int voted_for)
{
ha_state_t *ha = (ha_state_t *)ctx;
cluster_t *c = ha->raft_c;
/* Store ha-term.bin under {workdir}/{myname}/data/ alongside WAL/snapshot. */
char data_dir[1024] = {0};
if (c && c->myconf && c->myconf->workdir && c->myconf->name)
snprintf(data_dir, sizeof(data_dir), "%s/%s/data",
c->myconf->workdir, c->myconf->name);
const char *workdir = data_dir[0] ? data_dir : NULL;
/* Update ha->term */
atomic_store_explicit(&ha->term, term, memory_order_release);
/* Map voted_for index to name */
pthread_mutex_lock(&ha->lock);
if (voted_for == -1) {
ha->voted_for_term = 0;
memset(ha->voted_for_name, 0, sizeof(ha->voted_for_name));
} else {
ha->voted_for_term = term;
/* Look up candidate name by index */
if (c && c->myconf &&
voted_for < c->myconf->peers_cfg.candidates_count) {
const char *name = c->myconf->peers_cfg.candidates[voted_for].name;
strncpy(ha->voted_for_name, name, sizeof(ha->voted_for_name) - 1);
ha->voted_for_name[sizeof(ha->voted_for_name) - 1] = '\0';
} else {
/* Test mode or out-of-range: use "self" placeholder */
if (voted_for == ha->my_priority) {
strncpy(ha->voted_for_name, "self",
sizeof(ha->voted_for_name) - 1);
} else {
memset(ha->voted_for_name, 0, sizeof(ha->voted_for_name));
}
}
}
pthread_mutex_unlock(&ha->lock);
ha_persist_term(ha, workdir);
return 0;
}
/* raft_cb_trace: emit raft state-machine events at DEBUG3 level.
* Active when controller is started with -vvv; suppressed otherwise. */
static void raft_cb_trace(void *ctx, const char *event, const char *detail)
{
ha_state_t *ha = (ha_state_t *)ctx;
tlog(TLOG_DEBUG3, "RAFT[%s] term=%" PRIu64 " event=%s%s%s",
ha->my_priority == 0 ? "p0" :
ha->my_priority == 1 ? "p1" : "p2",
atomic_load_explicit(&ha->term, memory_order_relaxed),
event,
detail ? " " : "", detail ? detail : "");
}
static const raft_callbacks_t s_real_cbs = {
.on_send_vote_request = raft_cb_send_vote_request,
.on_vote_eligible = raft_cb_vote_eligible,
.on_become_leader = raft_cb_become_leader,
.on_become_follower = raft_cb_become_follower,
.on_persist = raft_cb_persist,
.on_schedule_timeout = raft_cb_schedule_timeout,
.on_cancel_timeout = raft_cb_cancel_timeout,
.on_start_election_after_delay = raft_cb_start_election_after_delay,
.on_trace = raft_cb_trace,
};
/* ── 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);
/* failover_requested / watchdog_stop deleted — watchdog replaced by ha_arm_heartbeat_timeout_vtask */
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);
/* ha-heartbeat-liveness Task 1.1: zero-init liveness timestamp.
* Written by ha_on_heartbeat_received (task 2.1); read by liveness_poll_fn. */
ha->last_activity = 0;
/* ha-repl-hardening Task 1.4: per-standby convergence tracking arrays.
* memset(ha, 0, sizeof(*ha)) above already zeroes these arrays; the explicit
* note documents the invariant: on entry to CTRL_PRIMARY_COMMITTING every
* standby starts as not-converged (tag=0, lsn=0, converged=false). */
/* 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);
/* ── Task 6.1/6.2: Initialize embedded raft_state_t ─────────────────────
*
* Wire real callbacks (s_real_cbs) so the state machine is ready when
* ha_start_election_async re-initialises raft with the correct n_peers /
* self_idx / initial_term after ha_restore_term().
*
* n_peers = 1 and self_idx = 0 are safe placeholders here; they are
* overwritten by ha_raft_reinit() (called from ha_start_election_async)
* which runs after ha_restore_term() and knows the real cluster size.
*
* initial_term = 0 / initial_voted_for = -1 are first-boot defaults;
* ha_raft_reinit() restores the persisted values before any election fires.
*
* ha->raft_c is NULL until ha_start_election_async sets it.
*/
ha->raft_c = NULL;
ha->raft_election_timeout_ctx = NULL;
raft_config_t raft_cfg;
memset(&raft_cfg, 0, sizeof(raft_cfg));
raft_cfg.n_peers = 1; /* placeholder; overwritten by ha_raft_reinit */
raft_cfg.self_idx = my_priority; /* self_idx == priority by convention */
raft_cfg.election_timeout_ms = cfg->election_timeout_ms;
raft_cfg.my_priority = my_priority;
raft_cfg.initial_term = 0; /* ha_restore_term() updates ha->term after ha_init */
raft_cfg.initial_voted_for = -1; /* no vote on fresh init */
raft_cfg.cbs = &s_real_cbs;
raft_cfg.ctx = ha;
raft_init(&ha->raft, &raft_cfg);
return 0;
}
/* SIGUSR1 handler / ha_install_sigusr1() deleted — failover_requested replaced by
* ha_arm_heartbeat_timeout_vtask (delaymgr) in election-iomgr-refactor */
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));
}
/* probe_thread() / wait_any_probe() deleted — only called by am_i_primary() (election-iomgr-refactor) */
/* am_i_primary() deleted — replaced by ha_start_election_async (election-iomgr-refactor) */
/*
* 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() deleted — never called (election-iomgr-refactor) */
/* ── 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() / ha_watchdog_thread() / ha_start_watchdog() /
* ha_stop_watchdog() deleted — watchdog pthread replaced by
* ha_arm_heartbeat_timeout_vtask (delaymgr) in election-iomgr-refactor */
void ha_fini(ha_state_t *ha)
{
if (!ha)
return;
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;
}
}
/* ── Task 2.1/2.2: WAL catch-up entry application (standby side) ─────────────
*
* ha_apply_wal_catchup_entry: apply WAL catch-up entries received from the
* primary to local persistent storage and advance applied_lsn on success.
*
* Two modes (Req 5.3, 5.5):
*
* Batch mode (entry->n_entries > 0):
* Iterate entry->entries[0..n_entries-1] in order. Apply each sub-entry
* individually via apply_wal_entry(). NULL sub-entries are treated as
* corrupt and skipped (Req 3.3). After ALL sub-entries are processed,
* advance applied_lsn to last-successfully-applied lsn+1 so the next
* heartbeat reply reflects post-batch state (Req 5.3).
*
* Single-entry mode (entry->n_entries == 0):
* Backward-compatible path from task 2.1. Apply entry->lsn/entry_type
* directly and advance applied_lsn to lsn+1 on success.
*
* In both modes:
* - applied_lsn is advanced atomically and monotonically.
* - On error for any individual entry: log ERROR, skip, continue (Req 3.3).
* - applied_lsn is in-memory only; not persisted to disk (design §Durability).
* - Standby stays in CTRL_STANDBY throughout (Req 3.4).
*
* Returns 0 on success (even if some batch entries were skipped due to errors).
* Returns -EINVAL if ha, p, or entry is NULL.
*/
/* ha_advance_applied_lsn: atomically advance ha->applied_lsn to new_lsn
* if new_lsn > current value (monotonic). */
static void ha_advance_applied_lsn(ha_state_t *ha, uint64_t new_lsn)
{
uint64_t old_lsn = (uint64_t)atomic_load_explicit(&ha->applied_lsn,
memory_order_acquire);
while (new_lsn > old_lsn) {
if (atomic_compare_exchange_weak_explicit(
&ha->applied_lsn, &old_lsn, new_lsn,
memory_order_acq_rel, memory_order_acquire))
break;
}
}
/* ha_apply_single_entry: apply one decoded wal_catchup_entry_t sub-entry.
* Returns 0 on success, non-zero on error (caller logs and skips). */
static int ha_apply_single_entry(persist_t *p, wal_catchup_entry_t *e)
{
wal_entry_hdr_t hdr;
memset(&hdr, 0, sizeof(hdr));
hdr.lsn = e->lsn;
hdr.type = (uint32_t)e->entry_type;
hdr.payload_len = 0;
hdr.crc = 0;
hdr.prev_crc = 0;
return apply_wal_entry(p, &hdr, NULL);
}
int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
wal_catchup_entry_t *entry)
{
if (!ha || !p || !entry)
return -EINVAL;
/* ── Batch mode: entry->n_entries > 0 ─────────────────────────────────── */
if (entry->n_entries > 0) {
/* Apply each sub-entry in LSN order. Track the last successfully
* applied LSN so we can advance applied_lsn after the full batch.
* Req 5.5: applied_lsn advances only after the full batch is written.
* Req 3.3: NULL or failed entries are logged and skipped; no crash. */
uint64_t last_good_lsn = 0;
int had_good = 0;
for (int i = 0; i < entry->n_entries; i++) {
wal_catchup_entry_t *sub = entry->entries[i];
if (!sub) {
ERROR("ha_apply_wal_catchup_entry: NULL sub-entry at "
"batch index %d — skipping", i);
continue;
}
int rc = ha_apply_single_entry(p, sub);
if (rc != 0) {
ERROR("ha_apply_wal_catchup_entry: apply_wal_entry failed "
"for batch index %d lsn=%" PRIu64 " type=%u rc=%d "
"— skipping entry",
i, sub->lsn, (unsigned)sub->entry_type, rc);
continue;
}
last_good_lsn = sub->lsn;
had_good = 1;
}
/* Advance applied_lsn to last_good_lsn+1 after the full batch.
* Only advance if at least one entry was applied successfully. */
if (had_good)
ha_advance_applied_lsn(ha, last_good_lsn + 1);
return 0;
}
/* ── Single-entry mode (task 2.1 backward-compatible path) ────────────── */
int rc = ha_apply_single_entry(p, entry);
if (rc != 0) {
ERROR("ha_apply_wal_catchup_entry: apply_wal_entry failed for "
"lsn=%" PRIu64 " type=%u rc=%d — skipping entry",
entry->lsn, (unsigned)entry->entry_type, rc);
return rc;
}
/* Advance applied_lsn atomically (lsn+1 mirrors persist->current_lsn
* after a successful apply). */
ha_advance_applied_lsn(ha, entry->lsn + 1);
return 0;
}
/* ha_nop_reply_state_report — populate *sr from ha state and wire into reply.
* Reads ha->applied_lsn (atomic) and myconf->config_version_tag, stores them
* in *sr, then sets reply_out->data_which = stateReportMsg and
* reply_out->data.state_report = sr.
* Caller keeps *sr alive until xce_reply() serialises the message.
* Requirements: 1.1, 1.3, 1.4 (ha-repl-hardening spec) */
void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf,
state_report_t *sr, xce_reply_t *reply_out)
{
sr->applied_lsn = atomic_load_explicit(&ha->applied_lsn,
memory_order_acquire);
sr->config_tag = myconf ? myconf->config_version_tag : 0;
reply_out->data_which = RequestReply_data_stateReportMsg;
reply_out->data.state_report = sr;
}
/*
* ha_process_standby_state — update per-standby convergence tracking.
*
* Called on each heartbeat reply from a standby (CTRL_PRIMARY_COMMITTING or
* CTRL_PRIMARY). Stores (reported_tag, reported_lsn) into the per-standby
* arrays, evaluates convergence, and logs INFO on the first
* not-converged → converged transition (Req 7.2). Regression resets the flag
* (Req 2.5).
*
* Convergence: reported_tag >= committed_tag AND reported_lsn >= committed_lsn.
*
* Requirements: 1.2, 2.2, 2.5 (ha-repl-hardening spec)
*/
void ha_process_standby_state(ha_state_t *ha, int peer_idx,
uint64_t reported_tag, uint64_t reported_lsn,
uint64_t committed_tag, uint64_t committed_lsn)
{
if (peer_idx < 0 || peer_idx >= MAX_HA_CANDIDATES)
return;
/* Store reported values */
ha->standby_reported_tag[peer_idx] = reported_tag;
ha->standby_reported_lsn[peer_idx] = reported_lsn;
/* Evaluate convergence */
bool was_converged = ha->standby_converged[peer_idx];
bool now_converged = (reported_tag >= committed_tag) &&
(reported_lsn >= committed_lsn);
ha->standby_converged[peer_idx] = now_converged;
/* Log INFO on first transition to converged (Req 7.2) */
if (now_converged && !was_converged) {
INFO("HA: standby[%d] converged: config_tag=%" PRIu64
" applied_lsn=%" PRIu64 " (committed_tag=%" PRIu64
" committed_lsn=%" PRIu64 ")",
peer_idx, reported_tag, reported_lsn,
committed_tag, committed_lsn);
}
}
/* ── 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;
}
/* Arm periodic heartbeat to standbys — keeps their watchdog alive when
* the cluster is idle (no WAL entries or config pushes). */
ha_start_primary_heartbeat(c);
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 2.3: higher term → step down BEFORE any role gate.
*
* Raft §5.1: "If RPC request or response contains term T > currentTerm:
* set currentTerm = T, convert to follower."
*
* Without this ordering, a node in CTRL_PRIMARY_COMMITTING at term=N
* would deny a vote request at term=N+1 without ever seeing the higher
* term. Both nodes then independently promote to PRIMARY via their
* committing timeout → split-brain. */
uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire);
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));
/* Step down from leader/committing: a higher-term election is in
* progress and we must not block it with the role gate below. */
ctrl_role_t cur_role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
if (cur_role == CTRL_PRIMARY || cur_role == CTRL_PRIMARY_COMMITTING) {
ha_do_demote(ha);
INFO("HA: stepping down (term %" PRIu64 " → %" PRIu64
") — higher-term election in progress", my_term, term);
}
my_term = term;
}
/* Req 2.2: stale term → deny, do not update any state */
if (term < my_term) {
return 0;
}
/* Req 6.1, 6.2: role-based deny gate.
* Evaluated AFTER the term catch-up above: if we just stepped down due to
* a higher term, role is now CTRL_STANDBY and we fall through to voting.
* PRIMARY and PRIMARY_COMMITTING at the current term still deny votes
* (protecting the live primary from same-term challengers). */
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;
/* Protect voted_for check+set from concurrent vote requests on different vtasks */
pthread_mutex_lock(&ha->lock);
/* 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) {
pthread_mutex_unlock(&ha->lock);
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) {
pthread_mutex_unlock(&ha->lock);
return 0;
}
if (config_tag == my_config_tag && last_lsn < my_last_lsn) {
pthread_mutex_unlock(&ha->lock);
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';
pthread_mutex_unlock(&ha->lock);
/* 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;
}
/* ── Test-only hook shims (XCE_TESTING only) ──────────────────────────────
* Type definitions (election_timeout_ctx_t, committing_timeout_ctx_t),
* extern declarations (cluster_init_phase2, push_config_to_standbys,
* ha_send_ctrlhello_to_all_standbys), and static forward declarations
* (committing_timeout_fn, ha_heartbeat_is_fresh) have been moved to the
* forward-declarations block near the top of this file so that the raft
* callback implementations (Task 6.2) can reference them.
*/
#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;
/*
* ha_nop_req_hook: test-only seam for xce_nop_req in primary_heartbeat_fn.
* When non-NULL, primary_heartbeat_fn calls this INSTEAD OF xce_nop_req so
* tests can count NOP sends without a live endpoint. Req 3.2, task 2.2.
*/
int (*ha_nop_req_hook)(endpoint_t *ep) = 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) {
/* Cancel any prior committing timeout */
committing_timeout_ctx_t *prev = (committing_timeout_ctx_t *)
atomic_exchange(&ha->committing_timeout_ctx, NULL);
if (prev) {
atomic_store(&prev->cancelled, 1);
}
committing_timeout_ctx_t *ct = calloc(1, sizeof(*ct));
if (ct) {
atomic_init(&ct->cancelled, 0);
ct->ha = ha;
ct->c = c;
atomic_store(&ha->committing_timeout_ctx, ct);
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);
atomic_store(&ha->committing_timeout_ctx, NULL);
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);
/* Primary connects outbound and pushes config via ctrlhello.
* Standby is fully passive here — no fetch_config_from_primary. */
}
}
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).
*/
/* vote_ep_ctx_t: typedef moved to Task 6.2 forward-declaration block above */
/*
* peer_ctrl_msg_handler / vote_reply_decode_and_dispatch /
* repl_ack_decode_and_dispatch: forward declarations.
* peer_ctrl_msg_handler is the single iomgr handler set in on_vote_connected
* and never reassigned — it routes by ha->role at dispatch time.
*/
static int peer_ctrl_msg_handler(void *ep_raw);
static int vote_reply_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx);
static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx);
#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
/* Set state-aware dispatcher — routes by ha->role at message-arrival time.
* Never reassigned: ELECTING → vote reply; COMMITTING/PRIMARY → repl ack. */
ep->msg_handler = peer_ctrl_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 — delegates to the raft library or
* the test hook (when XCE_TESTING and hook is non-NULL).
*
* Production path: raft_on_vote_reply owns quorum tracking and fires
* on_become_leader/on_become_follower via the raft callbacks.
* Test path: xce_vote_reply_hook captures decoded fields without real iomgr.
*/
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
(void)ctx; (void)peer_tag; (void)peer_lsn; /* consumed by raft library via callbacks */
raft_on_vote_reply(&ha->raft, idx, vote_granted, term);
}
/*
* vote_reply_decode_and_dispatch: decode a RequestVoteReply packet and dispatch
* to vote_reply_dispatch. Factored from vote_reply_msg_handler so that
* peer_ctrl_msg_handler can invoke the same logic without code duplication.
* Packet ownership: caller passes ep->fsm_in.pkt; this function frees it.
*/
static int vote_reply_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
{
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;
int ok = 0;
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;
}
memset(&vr_raw, 0, sizeof(vr_raw));
read_RequestVoteReply(&vr_raw, rep_s.data.requestvotereplymsg);
capn_free(&capn_ctx);
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:
if (pkt) { free_packet(pkt); free(pkt); }
if (ok) {
uint64_t cur = atomic_load_explicit(&ha->term, memory_order_acquire);
while (vr_term > cur) {
if (atomic_compare_exchange_weak_explicit(&ha->term, &cur, vr_term,
memory_order_acq_rel,
memory_order_acquire))
break;
}
vote_reply_dispatch(ha, ctx, idx, vr_term, vr_granted, vr_tag, vr_lsn);
} else {
vote_reply_dispatch(ha, ctx, idx,
atomic_load_explicit(&ha->term, memory_order_acquire),
0, 0, 0);
}
return 0;
}
/*
* repl_ack_decode_and_dispatch: decode an inbound packet on a promoted vote
* endpoint while role == CTRL_PRIMARY_COMMITTING or CTRL_PRIMARY.
*
* - OPCode_replCatchUpDone → ha_repl_ack_received
* - OPCode_reply with stateReportMsg → ha_process_standby_state (Task 3.1)
* - OPCode_nop or OPCode_reply without stateReport → silently ignored
* - anything else → WARN + drop
*/
static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
{
ha_state_t *ha = vctx->ha;
cluster_t *c = vctx->c;
io_packet_t *pkt = ep->fsm_in.pkt;
if (!pkt || !pkt->data || pkt->sz == 0) {
if (pkt) { free_packet(pkt); free(pkt); }
return 0;
}
struct capn capn_ctx;
struct Request outer_rq;
Request_ptr outer_rp;
int capn_ok = capn_init_mem(&capn_ctx, (const uint8_t *)pkt->data,
(size_t)pkt->sz, 0);
free_packet(pkt);
free(pkt);
ep->fsm_in.pkt = NULL;
if (capn_ok != 0)
return 0;
outer_rp.p = capn_getp(capn_root(&capn_ctx), 0, 1);
read_Request(&outer_rq, outer_rp);
int opcode = (int)outer_rq.opcode;
if (opcode == OPCode_replCatchUpDone) {
capn_free(&capn_ctx);
ha_repl_ack_received(ha, c);
} else if (opcode == OPCode_reply &&
outer_rq.body_which == Request_body_replymsg &&
outer_rq.body.replymsg.p.data) {
/* Heartbeat NOP reply — decode inner RequestReply for stateReport */
struct RequestReply rep_s;
memset(&rep_s, 0, sizeof(rep_s));
read_RequestReply(&rep_s, outer_rq.body.replymsg);
if (rep_s.data_which == RequestReply_data_stateReportMsg) {
/* Decode stateReport and call convergence tracker (Task 3.1).
* read_StateReport reads scalar fields directly from the capnp
* pointer without heap allocation — safe to call before capn_free. */
struct StateReport sr_raw;
memset(&sr_raw, 0, sizeof(sr_raw));
read_StateReport(&sr_raw, rep_s.data.stateReportMsg);
state_report_t sr;
sr.applied_lsn = (uint64_t)sr_raw.appliedLsn;
sr.config_tag = (uint64_t)sr_raw.configTag;
capn_free(&capn_ctx);
/* peer_idx from standby_ep_ctx_t (layout-compatible with vote_ep_ctx_t) */
int peer_idx = vctx->peer_idx;
uint64_t committed_lsn = 0;
uint64_t committed_tag = 0;
#ifndef XCE_TESTING
if (c) {
committed_lsn = atomic_load_explicit(&c->persist.committed_lsn,
memory_order_acquire);
committed_tag = atomic_load_explicit(&ha->committed_config_tag,
memory_order_acquire);
}
#else
/* In test mode, c may be NULL; caller sets ha fields directly */
committed_lsn = atomic_load_explicit(&ha->last_known_committed_lsn,
memory_order_acquire);
committed_tag = atomic_load_explicit(&ha->committed_config_tag,
memory_order_acquire);
(void)c;
#endif
ha_process_standby_state(ha, peer_idx,
sr.config_tag, sr.applied_lsn,
committed_tag, committed_lsn);
} else {
capn_free(&capn_ctx);
}
} else if (opcode == OPCode_nop) {
/* Plain NOP with no body — silently ignore */
capn_free(&capn_ctx);
} else {
capn_free(&capn_ctx);
WARN("repl_ack_decode_and_dispatch: unexpected opcode %d on promoted ep",
opcode);
}
return 0;
}
/*
* peer_ctrl_msg_handler: state-aware iomgr handler set once in on_vote_connected.
* Routes by ha->role at dispatch time — no explicit handler reassignment needed.
* CTRL_ELECTING → vote_reply_decode_and_dispatch
* CTRL_PRIMARY_COMMITTING /
* CTRL_PRIMARY → repl_ack_decode_and_dispatch
* any other role → drop packet (WARN)
*/
static int peer_ctrl_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;
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
switch (role) {
case CTRL_ELECTING:
return vote_reply_decode_and_dispatch(ep, vctx);
case CTRL_PRIMARY_COMMITTING:
case CTRL_PRIMARY:
return repl_ack_decode_and_dispatch(ep, vctx);
default: {
io_packet_t *pkt = ep->fsm_in.pkt;
if (pkt) { free_packet(pkt); free(pkt); ep->fsm_in.pkt = NULL; }
WARN("peer_ctrl_msg_handler: unexpected role %d, packet dropped", (int)role);
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_for_test 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: test entry point for vote-reply decode path.
* Temporarily forces role to CTRL_ELECTING so peer_ctrl_msg_handler routes to
* vote_reply_decode_and_dispatch, then restores the original role.
* 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)
{
vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data;
ha_state_t *ha = vctx ? vctx->ha : NULL;
int32_t saved_role = ha ? atomic_load_explicit(&ha->role, memory_order_acquire) : 0;
if (ha)
atomic_store_explicit(&ha->role, (int32_t)CTRL_ELECTING, memory_order_release);
ep->fsm_in.pkt = pkt;
int rc = peer_ctrl_msg_handler(ep);
if (ha)
atomic_store_explicit(&ha->role, saved_role, memory_order_release);
return rc;
}
/*
* 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;
}
/*
* repl_catchup_done_pkt_for_test: build a minimal io_packet_t encoding
* Request{opcode=replCatchUpDone, body_which=nodata}.
* Used by peer_ctrl_msg_handler tests to verify the COMMITTING dispatch path.
*/
io_packet_t *repl_catchup_done_pkt_for_test(void)
{
struct capn cs;
struct capn_segment *seg;
struct Request rq_capn;
Request_ptr rq_ptr;
io_packet_t *pkt;
int rc;
capn_init_malloc(&cs);
seg = capn_root(&cs).seg;
memset(&rq_capn, 0, sizeof(rq_capn));
rq_capn.opcode = OPCode_replCatchUpDone;
rq_capn.seq = 1;
rq_capn.body_which = Request_body_nomsg;
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_ep_err_handler_for_test: call the static vote_ep_err_handler on ep.
* Allows tests to invoke the err_handler without going through iomgr/endpoint machinery.
*/
int vote_ep_err_handler_for_test(endpoint_t *ep)
{
return vote_ep_err_handler((void *)ep);
}
/*
* peer_ctrl_msg_handler_for_test: expose peer_ctrl_msg_handler for tests.
* Sets ep->fsm_in.pkt then invokes the handler. Packet ownership transferred.
*/
int peer_ctrl_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt)
{
ep->fsm_in.pkt = pkt;
return peer_ctrl_msg_handler(ep);
}
/*
* vote_ep_ctx_set_cluster_for_test: set the cluster_t pointer inside the
* vote_ep_ctx_t stored at ep->data. Tests that need ha_repl_ack_received
* to call ha_promote_to_primary(c) when quorum is reached must call this
* after vote_ep_ctx_build_for_test (which sets vctx->c = NULL by default).
* Requirements: task 7.3 (ha-peer-connection-reuse).
*/
void vote_ep_ctx_set_cluster_for_test(endpoint_t *ep, cluster_t *c)
{
if (!ep || !ep->data) return;
vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data;
vctx->c = c;
}
/*
* raft_cb_become_leader_for_test: drive the leader promotion path without
* a live cluster. Passes ha as ctx; sets ha->raft_c = NULL to suppress
* production-only calls (phase2, push_config, ctrlhello, committing timer).
*/
void raft_cb_become_leader_for_test(ha_state_t *ha, uint64_t term)
{
ha->raft_c = NULL;
raft_cb_become_leader((void *)ha, term);
}
/*
* raft_cb_become_follower_for_test: drive the follower path without a live
* cluster. Passes ha as ctx; sets ha->raft_c = NULL to suppress production
* calls (phase2, fetch_config); ha_arm_heartbeat_timeout_vtask is a no-op
* when c=NULL so no watchdog thread is spawned.
*/
void raft_cb_become_follower_for_test(ha_state_t *ha, uint64_t term)
{
ha->raft_c = NULL;
raft_cb_become_follower((void *)ha, term, -1);
}
#endif /* XCE_TESTING */
/* ── 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;
ha_state_t *ha = vctx->ha;
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
if (role == CTRL_PRIMARY_COMMITTING || role == CTRL_PRIMARY) {
/* Promoted connection dropped — remove from standby_endpoints */
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
for (int i = 0; i < ha->standby_endpoints->count; i++) {
if (zarray_get(ha->standby_endpoints, i) == ep) {
zarray_del(ha->standby_endpoints, i);
break;
}
}
pthread_mutex_unlock(&ha->lock);
}
} else {
/* Election phase: count as denied vote */
handle_request_vote_reply(ha, vctx->ctx, vctx->peer_idx,
vctx->my_term,
0, /* vote_granted = false */
0, 0);
}
/* Do NOT free vctx here — peer_ctrl_msg_handler may still reference ep->data.
* vctx is freed in ha_election_cleanup() when all per-peer contexts release. */
return 0;
}
/* ── Task 6.3: ha_start_election_async — delegate to raft_start_election ────
*
* Non-blocking election initiator. Sets up the raft state machine and
* delegates all election mechanics to raft_start_election().
*
* raft_start_election() handles:
* - term increment and self-vote (via on_persist callback)
* - single-node shortcircuit (fires on_become_leader immediately)
* - multi-node RPC fan-out (via on_send_vote_request callback per peer)
* - election timeout scheduling (via on_schedule_timeout callback)
*
* ha_elect_ctx_t is still allocated here because raft_cb_send_vote_request
* reads ha->elect_ctx to retrieve my_tag/my_lsn for the capnp VoteRequest
* payload. Vote counting and quorum decisions are now owned by the raft
* library (raft_on_vote_reply fires on_become_leader/follower callbacks).
*
* Requirements: 1.3, 2.1 (shortcircuit), 2.4 (async vote send), 9.2
*/
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)
{
(void)peers; /* peer connections driven by raft_cb_send_vote_request */
(void)my_name; /* self-vote name stored in on_persist via raft_start_election */
/* Step 0: wire cluster pointer for raft callbacks.
* Must be set before any raft_* call so callbacks can reach cluster_t. */
ha->raft_c = c;
/* Step 0b: re-initialise raft_state_t with the real cluster size and
* the term/voted_for restored from disk by ha_restore_term().
*
* ha->raft.term may be 0 from ha_init while ha->term was restored to a
* higher value by ha_restore_term() — reconcile here before the election.
*
* self_idx = ha->my_priority (index of this node in candidates[]).
* n_peers = total cluster size (candidates_count); passed directly.
*
* voted_for: map ha->voted_for_name back to a peer index (-1 if none).
*/
{
int restored_voted_for = -1;
if (ha->voted_for_name[0] != '\0' && c && c->myconf) {
const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
for (int i = 0; i < pcfg->candidates_count; i++) {
if (strcmp(pcfg->candidates[i].name, ha->voted_for_name) == 0) {
restored_voted_for = i;
break;
}
}
}
raft_config_t rcfg;
memset(&rcfg, 0, sizeof(rcfg));
rcfg.n_peers = n_peers; /* total candidates_count */
rcfg.self_idx = ha->my_priority;
rcfg.election_timeout_ms = ha->config.election_timeout_ms;
rcfg.my_priority = ha->my_priority;
rcfg.initial_term = atomic_load_explicit(&ha->term, memory_order_acquire);
rcfg.initial_voted_for = restored_voted_for;
rcfg.cbs = &s_real_cbs;
rcfg.ctx = ha;
raft_init(&ha->raft, &rcfg);
}
/* Step 1: store peer count for ha_repl_ack_received quorum calculation */
ha->n_peers = n_peers;
/* Step 2: transition to CTRL_ELECTING before firing RPCs.
* raft_start_election fires on_become_leader / on_send_vote_request below;
* the role must already be ELECTING so ha_dispatch_gate() admits vote RPCs. */
atomic_store_explicit(&ha->role, (int32_t)CTRL_ELECTING, memory_order_release);
INFO("HA FSM: → CTRL_ELECTING (async, n_peers=%d)", n_peers);
/* Step 3: pre-allocate ha_elect_ctx_t so raft_cb_send_vote_request can
* read my_tag/my_lsn from it when building the capnp VoteRequest payload.
* Vote counting is now owned by the raft library; this ctx is only a
* metadata carrier for tag/lsn. For multi-node elections it is kept alive
* until all per-peer vote_ep_ctx_t endpoints finish (iomgr lifetime).
* For single-node, raft_start_election fires on_become_leader immediately
* so no ctx reads occur after return. */
int n_other = (n_peers > 1) ? n_peers - 1 : 0;
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");
return;
}
ctx->n_peers = n_other;
ctx->votes_granted = 1; /* self-vote placeholder (raft tracks actual count) */
ctx->n_replied = 0;
ctx->max_resp_tag = 0;
ctx->max_resp_lsn = 0;
ctx->result = -1;
ctx->done_posted = 0;
ctx->c = c;
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
/* Store candidate's own tag/lsn; votes[0] is read by raft_cb_send_vote_request */
if (n_other > 0) {
ctx->votes[0].my_tag = my_config_tag;
ctx->votes[0].my_lsn = my_lsn;
}
/* Store ctx in ha so raft_cb_send_vote_request can find it via ha->elect_ctx */
atomic_store_explicit(&ha->elect_ctx, ctx, memory_order_release);
/* Step 4: delegate election to raft library.
*
* A node with no valid local config (my_config_tag == 0) must not start an
* election — any peer with a real config tag will beat it in the freshness
* gate and the node would waste a term increment. It stays in CTRL_ELECTING,
* accepts incoming vote requests (granting to fresher peers), and will
* become STANDBY after the winner sends it a heartbeat/ctrlhello.
* n_peers and raft state are already initialised above so RDHUP retries work.
*
* raft_start_election when allowed:
* - increments ha->raft.term and records self-vote (fires on_persist)
* - single-node: fires on_become_leader → raft_cb_become_leader
* - multi-node: fires on_send_vote_request per peer → raft_cb_send_vote_request
* then fires on_schedule_timeout → raft_cb_schedule_timeout
*
* No direct ha->term++ or ha->voted_for_* writes happen in this function
* after this call; the raft library owns those via on_persist. */
/* Req 4.3 / ha-config-bootstrap Req 2.8:
* A node with no local config (my_config_tag == 0) must not start an
* election if any peer has a valid config — it cannot win (freshness gate
* denies it) and it would waste a term increment. It stays in CTRL_ELECTING,
* accepts incoming vote requests (granting to fresher peers), and becomes
* STANDBY after the winner sends a heartbeat.
*
* Exception: if ALL peers also have config_tag=0 (cluster cold-start with no
* configured node), someone must win — allow election normally.
*
* my_config_tag comes from myconf->config_version_tag set by load_primary_config;
* n_peers and the peer list from --peers are always available via the parameters.
* We check if any peer returned a non-zero tag via the xce_probe_config_tag path
* already called in myapp_run before ha_start_election_async. */
if (my_config_tag == 0 && n_peers > 1) {
INFO("ha_start_election_async: no local config (tag=0, n_peers=%d) — "
"participating as voter only, not starting own election", n_peers);
/* Transition to CTRL_STANDBY so myapp_run schedules fetch_config_from_primary.
* The primary will push config via ctrlhello once it enters COMMITTING. */
atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY, memory_order_release);
/* Free ctx — no vote sends will happen from this node this round. */
atomic_store_explicit(&ha->elect_ctx, NULL, memory_order_release);
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
free(ctx);
return;
}
raft_start_election(&ha->raft);
/* For single-node, raft_start_election fires on_become_leader synchronously.
* on_become_leader (raft_cb_become_leader) handles role → COMMITTING,
* phase2, and committing timeout. Free the ctx here since no raft callbacks
* will reference it after raft_start_election returns for single-node. */
if (n_peers == 1) {
atomic_store_explicit(&ha->elect_ctx, NULL, memory_order_release);
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
free(ctx);
return;
}
INFO("ha_start_election_async: returning (n_peers=%d, raft election started)",
n_peers);
/* Returns immediately — election result delivered via raft on_become_leader/follower */
}
#ifdef XCE_TESTING
/*
* 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
/* ── 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;
/* Cancel the liveness poll task first (Req 5.4): prevents a concurrent
* poll fn from triggering a second failover call after EPOLLRDHUP fires. */
ha_disarm_heartbeat_timeout_vtask(ha);
/* 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);
/* Task 6.3: delegate RDHUP election trigger to raft library.
*
* raft_on_leader_lost:
* - clears ha->raft.voted_for = -1 (no direct ha->voted_for_* writes here)
* - sets ha->raft.role = RAFT_FOLLOWER
* - fires on_start_election_after_delay(ctx, priority * election_timeout_ms)
* → raft_cb_start_election_after_delay schedules raft_rdhup_delay_fn
* which calls raft_check_and_start_election after the priority jitter.
*
* The manual nanosleep jitter + voted_for clear + skip-if-already-voted
* logic is now fully owned by the raft state machine.
*
* raft->raft_c must be set before raft_on_leader_lost fires callbacks. */
ha->raft_c = c;
#ifdef XCE_TESTING
if (ha_rdhup_election_hook != NULL) {
/* Test seam: build peer args for hook (same as before) so existing
* tests that capture ha_start_election_async arguments still work. */
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;
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 : "";
ha_rdhup_election_hook(ha, c, np, peer_ptrs, my_tag, my_lsn, my_name);
}
return;
}
#endif
raft_on_leader_lost(&ha->raft);
}
/* ── Standby liveness poll (replaces hb_timeout_thread) ─────────────────── */
/*
* liveness_poll_ctx_t: context for the delaymgr-based standby liveness task.
*
* Requirements: 1.2, 1.4, 1.5, 1.6, 5.4
*
* Replaces hb_vtask_t + hb_timeout_thread. No pthread, no semaphore.
* The poll fn reads ha->last_activity (written by ha_on_heartbeat_received)
* and fires ha_on_primary_ctrl_fd_rdhup if elapsed >= heartbeat_timeout_ms.
*
* cancelled is _Atomic so that writes from ha_disarm_heartbeat_timeout_vtask
* (main/vtask context) are visible to liveness_poll_fn (delaymgr worker thread)
* without a data race.
*/
typedef struct {
ha_state_t *ha;
cluster_t *c;
_Atomic int cancelled;
} liveness_poll_ctx_t;
/* Static pointer to the active context. _Atomic so concurrent arm/disarm
* vs poll-fn execution is well-defined. Access pattern:
* arm — allocates and stores (single writer, vtask context)
* disarm — sets cancelled, frees (single writer, vtask context)
* poll_fn — reads via passed arg only; never dereferences s_liveness_ctx
*/
static liveness_poll_ctx_t * _Atomic s_liveness_ctx = NULL;
/*
* liveness_poll_fn: delaymgr task replacing hb_timeout_thread.
*
* Checks whether the primary has been silent for >= heartbeat_timeout_ms.
* If so, calls ha_on_primary_ctrl_fd_rdhup and does not re-arm.
* If not, re-arms itself for the next poll interval.
* If cancelled, returns immediately without any action.
*
* Also skips if role != CTRL_STANDBY (Req 5.4): prevents duplicate failover
* if EPOLLRDHUP already transitioned the role to CTRL_ELECTING.
*/
static int liveness_poll_fn(void *arg)
{
liveness_poll_ctx_t *ctx = (liveness_poll_ctx_t *)arg;
/* Check cancelled flag first — set by ha_disarm_heartbeat_timeout_vtask */
if (atomic_load_explicit(&ctx->cancelled, memory_order_acquire))
return 0;
ha_state_t *ha = ctx->ha;
cluster_t *c = ctx->c;
/* Skip if not standby — EPOLLRDHUP may have already changed the role */
if (atomic_load_explicit(&ha->role, memory_order_acquire) != CTRL_STANDBY)
return 0;
/* Compute elapsed time since last primary message */
struct timespec now_ts;
clock_gettime(CLOCK_MONOTONIC, &now_ts);
time_t now = now_ts.tv_sec;
time_t last = ha->last_activity; /* time_t write is atomic on LP64 */
time_t elapsed = (last > 0) ? (now - last) : 0;
uint32_t timeout_s = ha->config.heartbeat_timeout_ms / 1000;
if (last > 0 && (uint32_t)elapsed >= timeout_s) {
/* Silence exceeded timeout — trigger failover */
WARN("HA: liveness poll: no primary activity for %lus (timeout=%us) → failover",
(unsigned long)elapsed, timeout_s);
ha_on_primary_ctrl_fd_rdhup(ha, c);
/* Do NOT re-arm — failover handles re-election */
return 0;
}
/* Still fresh — re-arm for next poll interval */
if (c && c->delaymgr.running) {
uint32_t poll_ms = ha->config.heartbeat_interval_ms / 2;
if (poll_ms < 100) poll_ms = 100;
delay_task_ms(&c->delaymgr, (int)poll_ms, 1,
(delay_task_entry_fn_t)liveness_poll_fn, ctx);
}
return 0;
}
#ifdef XCE_TESTING
/*
* liveness_poll_fn_test_entry: test-only entry point that calls liveness_poll_fn
* directly with a pre-constructed ctx, bypassing delaymgr.
* Only compiled when XCE_TESTING is defined.
*/
void liveness_poll_fn_test_entry(liveness_poll_ctx_t *ctx)
{
liveness_poll_fn(ctx);
}
/*
* primary_heartbeat_fn_for_test: call primary_heartbeat_fn directly with c.
* Bypasses delaymgr scheduling so tests can drive the idle-NOP logic without
* a live timer. Only compiled when XCE_TESTING is defined. (Task 2.2)
*/
int primary_heartbeat_fn_for_test(cluster_t *c)
{
return primary_heartbeat_fn(c);
}
#endif
void ha_arm_heartbeat_timeout_vtask(ha_state_t *ha, cluster_t *c)
{
if (!ha) return;
/* Disarm any existing poll ctx */
liveness_poll_ctx_t *old = atomic_load_explicit(&s_liveness_ctx,
memory_order_acquire);
if (old) {
atomic_store_explicit(&old->cancelled, 1, memory_order_release);
free(old);
atomic_store_explicit(&s_liveness_ctx, NULL, memory_order_release);
}
liveness_poll_ctx_t *ctx = calloc(1, sizeof(*ctx));
if (!ctx) return;
ctx->ha = ha;
ctx->c = c;
atomic_init(&ctx->cancelled, 0);
/* Initialise last_activity to now so the first poll computes elapsed from
* the arm time rather than from zero. */
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
ha->last_activity = ts.tv_sec;
atomic_store_explicit(&s_liveness_ctx, ctx, memory_order_release);
uint32_t poll_ms = ha->config.heartbeat_interval_ms / 2;
if (poll_ms < 100) poll_ms = 100;
INFO("HA: standby liveness poll armed (timeout=%ums, poll=%ums)",
ha->config.heartbeat_timeout_ms, poll_ms);
if (c && c->delaymgr.running) {
int rc = delay_task_ms(&c->delaymgr, (int)poll_ms, 1,
(delay_task_entry_fn_t)liveness_poll_fn, ctx);
if (rc < 0)
WARN("HA: ha_arm_heartbeat_timeout_vtask: delay_task_ms failed: %d", rc);
}
}
void ha_on_heartbeat_received(ha_state_t *ha)
{
if (!ha) return;
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
ha->last_activity = ts.tv_sec;
tlog(TLOG_DEBUG3, "HA: heartbeat received — last_activity updated to %ld", (long)ts.tv_sec);
}
void ha_disarm_heartbeat_timeout_vtask(ha_state_t *ha)
{
(void)ha;
liveness_poll_ctx_t *ctx = atomic_load_explicit(&s_liveness_ctx,
memory_order_acquire);
if (!ctx) return;
atomic_store_explicit(&ctx->cancelled, 1, memory_order_release);
free(ctx);
atomic_store_explicit(&s_liveness_ctx, NULL, memory_order_release);
}
/* ── Primary heartbeat sender + late-joiner outbound connect ─────────────── */
/*
* standby_ep_ctx_t: per-peer context for outbound standby connections opened
* by the primary during the heartbeat task.
*
* Layout deliberately matches vote_ep_ctx_t for the first 4 fields so that
* peer_ctrl_msg_handler (which casts ep->data to vote_ep_ctx_t*) can safely
* read ha, c, and peer_idx. ctx is always NULL here (no active election).
*/
typedef struct {
ha_state_t *ha;
cluster_t *c;
ha_elect_ctx_t *ctx; /* always NULL — not a vote endpoint */
int peer_idx; /* index into c->myconf->peers_cfg.candidates[] */
/* Task 1.1 (ha-heartbeat-liveness): last time a push/nop was sent to this
* standby (CLOCK_MONOTONIC seconds). Initialized to 0 at allocation;
* written by on_standby_connected and primary_heartbeat_fn (task 2.2). */
time_t last_push_time;
} standby_ep_ctx_t;
#ifdef XCE_TESTING
/*
* standby_ep_ctx_build_for_test: allocate a standby_ep_ctx_t, wire ep->data,
* and return ep. last_push_time is initialised to 0 (calloc).
* Caller must free(ep->data) after the test. (Task 2.2)
*/
endpoint_t *standby_ep_ctx_build_for_test(endpoint_t *ep,
ha_state_t *ha,
cluster_t *c,
int peer_idx)
{
standby_ep_ctx_t *sctx = calloc(1, sizeof(*sctx));
if (!sctx) return NULL;
sctx->ha = ha;
sctx->c = c;
sctx->ctx = NULL;
sctx->peer_idx = peer_idx;
/* last_push_time = 0 from calloc — tests set it explicitly */
ep->data = sctx;
return ep;
}
/*
* on_standby_connected_for_test: call on_standby_connected directly.
* ep->data must be a valid standby_ep_ctx_t (use standby_ep_ctx_build_for_test).
* xce_request will fail on a null fd — tests only check last_push_time. (Task 2.2)
*/
int on_standby_connected_for_test(endpoint_t *ep)
{
on_standby_connected(ep);
return 0;
}
#endif /* XCE_TESTING */
/*
* on_standby_connected: iomgr on_connected callback for outbound standby
* endpoints created by the heartbeat task.
*
* Sends OPCode_hello (CtrlHelloRequest) so the receiving standby runs
* handle_ctrlhello and triggers WAL catch-up + config push. Registers the
* endpoint in ha->standby_endpoints so future heartbeats reach it.
*/
static void on_standby_connected(endpoint_t *ep)
{
standby_ep_ctx_t *sctx = (standby_ep_ctx_t *)ep->data;
ha_state_t *ha = sctx->ha;
cluster_t *c = sctx->c;
/* Register in standby_endpoints under lock (idempotency guard) */
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
int already = 0;
for (int i = 0; i < ha->standby_endpoints->count; i++) {
if (zarray_get(ha->standby_endpoints, i) == ep) { already = 1; break; }
}
if (!already)
zarray_store(ha->standby_endpoints, ep);
pthread_mutex_unlock(&ha->lock);
}
/* Record last_push_time before any send so the first heartbeat tick after
* connect does not immediately send a redundant NOP (Req 3.3, task 2.2). */
{
struct timespec _ts;
clock_gettime(CLOCK_MONOTONIC, &_ts);
sctx->last_push_time = _ts.tv_sec;
}
/* Send ctrlhello — triggers handle_ctrlhello on the standby side */
uint64_t lsn = atomic_load_explicit(&c->persist.committed_lsn,
memory_order_acquire);
uint64_t tag = c->myconf ? c->myconf->config_version_tag : 0;
char *name = c->myconf ? c->myconf->name : "";
ctrl_hello_req_t hello = { .name = name, .key = NULL,
.applied_lsn = lsn, .config_tag = tag };
xce_request_t req;
memset(&req, 0, sizeof(req));
req.opcode = OPCode_hello;
req.body_which = Request_body_ctrlhellomsg;
req.body.ctrlhellomsg = &hello;
int rc = xce_request(ep, &req);
if (rc < 0)
WARN("HA: ctrlhello to peer[%d] failed: %d", sctx->peer_idx, rc);
else
INFO("HA: outbound standby connection to peer[%d] established, ctrlhello sent",
sctx->peer_idx);
/* Set state-aware dispatcher for subsequent messages on this endpoint */
ep->msg_handler = peer_ctrl_msg_handler;
}
/*
* standby_ep_err_handler_primary: error handler for outbound standby
* endpoints created by the heartbeat task. Removes from standby_endpoints
* and frees the context — the next heartbeat tick will attempt reconnect.
*/
static int standby_ep_err_handler_primary(void *ep_raw)
{
endpoint_t *ep = (endpoint_t *)ep_raw;
standby_ep_ctx_t *sctx = (standby_ep_ctx_t *)ep->data;
if (!sctx) return 0;
ha_state_t *ha = sctx->ha;
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
for (int i = 0; i < ha->standby_endpoints->count; i++) {
if (zarray_get(ha->standby_endpoints, i) == ep) {
zarray_del(ha->standby_endpoints, i);
break;
}
}
pthread_mutex_unlock(&ha->lock);
}
free(sctx);
ep->data = NULL;
return 0;
}
/*
* ha_connect_to_peer: open a new outbound TLS connection to a peer candidate
* and register it as a standby channel. Called from primary_heartbeat_fn
* for peers that are absent from standby_endpoints.
*/
static void ha_connect_to_peer(cluster_t *c, int peer_idx)
{
ha_state_t *ha = &c->ha;
const xce_candidate_t *peer =
&c->myconf->peers_cfg.candidates[peer_idx];
standby_ep_ctx_t *sctx = calloc(1, sizeof(*sctx));
if (!sctx) return;
sctx->ha = ha;
sctx->c = c;
sctx->peer_idx = peer_idx;
endpoint_t *ep = endpoint_get();
if (!ep) { free(sctx); return; }
endpoint_init(ep, AF_INET, SOCK_STREAM, 0, 0);
endpoint_set_peer(ep, c->myhost);
ep->data = sctx;
ep->iomgr = &c->iomgr;
if (ha->client_ssl_ctx) {
SSL_CTX *client_ctx = (SSL_CTX *)ha->client_ssl_ctx;
ep->ssl = SSL_new(client_ctx);
if (ep->ssl)
SSL_set_connect_state(ep->ssl);
}
endpoint_set_on_connected(ep, on_standby_connected);
endpoint_set_err_handler(ep, standby_ep_err_handler_primary);
int rc = endpoint_connect(ep, peer->hostname, peer->port);
if (rc < 0) {
WARN("HA: connect to peer[%d] (%s:%u) failed: %d",
peer_idx, peer->hostname, peer->port, rc);
endpoint_fini(ep);
free(sctx);
return;
}
rc = iomgr_monitor(ep);
if (rc < 0) {
endpoint_fini(ep);
free(sctx);
return;
}
INFO("HA: connecting outbound to peer[%d] (%s:%u)",
peer_idx, peer->hostname, peer->port);
}
/*
* primary_heartbeat_fn: self-perpetuating one-shot delaymgr task.
*
* On each tick:
* 1. Sends OPCode_nop to all connected standbys (keeps watchdog alive).
* 2. Scans the peers list; for any peer NOT in standby_endpoints, opens a
* fresh outbound TLS connection so late joiners are bootstrapped without
* the standby ever initiating an outbound connect.
*
* Stops naturally on demotion: the role check prevents re-arming.
*/
static int primary_heartbeat_fn(void *arg)
{
cluster_t *c = (cluster_t *)arg;
ha_state_t *ha = &c->ha;
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
if (role != CTRL_PRIMARY && role != CTRL_PRIMARY_COMMITTING)
return 0; /* demoted — stop the heartbeat loop */
/* 1. Heartbeat all connected standbys — idle-triggered NOP (Req 3.1-3.4).
*
* For each standby ep, compute elapsed time since the last outbound push.
* Send NOP only when elapsed >= heartbeat_interval_ms so that WAL entries,
* config pushes, and ctrlhello suppress redundant NOPs (Req 3.4).
*/
{
struct timespec _hb_ts;
clock_gettime(CLOCK_MONOTONIC, &_hb_ts);
time_t now = _hb_ts.tv_sec;
int interval_s = (int)(ha->config.heartbeat_interval_ms / 1000);
if (interval_s <= 0)
interval_s = HA_DEFAULT_HEARTBEAT_INTERVAL_MS / 1000;
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
int n = ha->standby_endpoints->count;
for (int i = 0; i < n; i++) {
endpoint_t *ep = (endpoint_t *)zarray_get(ha->standby_endpoints, i);
if (!ep) continue;
standby_ep_ctx_t *sctx = (standby_ep_ctx_t *)ep->data;
if (!sctx) continue;
time_t elapsed = now - sctx->last_push_time;
if (elapsed >= (time_t)interval_s) {
#ifdef XCE_TESTING
int rc = ha_nop_req_hook ? ha_nop_req_hook(ep) : xce_nop_req(ep);
#else
int rc = xce_nop_req(ep);
#endif
if (rc < 0)
WARN("HA: heartbeat nop to standby[%d] failed: %d", i, rc);
else
sctx->last_push_time = now;
}
}
pthread_mutex_unlock(&ha->lock);
}
}
/* 2. Reconnect to any peer missing from standby_endpoints */
if (c->myconf) {
const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
for (int i = 0; i < pcfg->candidates_count; i++) {
if (i == ha->my_priority) continue; /* skip self */
/* Check if this peer already has a standby endpoint.
*
* Two endpoint types can be in standby_endpoints:
* - Promoted vote eps: ep->data = vote_ep_ctx_t, peer_idx is
* votes_idx (relative, self excluded). Convert absolute i to
* votes_idx using the same formula as raft_cb_send_vote_request.
* - Heartbeat standby eps: ep->data = standby_ep_ctx_t (ctx==NULL),
* peer_idx is absolute candidate index. Compare directly with i.
*/
int votes_idx_for_i = (i < ha->my_priority) ? i : i - 1;
int found = 0;
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
for (int j = 0; j < ha->standby_endpoints->count; j++) {
endpoint_t *sep = (endpoint_t *)zarray_get(
ha->standby_endpoints, j);
if (!sep || !sep->data) continue;
standby_ep_ctx_t *sctx = (standby_ep_ctx_t *)sep->data;
int match_idx = (sctx->ctx != NULL)
? votes_idx_for_i /* promoted vote ep: relative */
: i; /* standby ep: absolute */
if (sctx->peer_idx == match_idx) { found = 1; break; }
}
pthread_mutex_unlock(&ha->lock);
}
if (!found) {
INFO("HA: peer[%d] (%s) not connected — initiating outbound connect",
i, pcfg->candidates[i].name ? pcfg->candidates[i].name : "?");
ha_connect_to_peer(c, i);
}
}
}
/* Re-arm for next interval — only reached when still PRIMARY */
int interval_ms = (int)ha->config.heartbeat_interval_ms;
if (interval_ms <= 0)
interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS;
delay_task_ms(&c->delaymgr, interval_ms, 1,
(delay_task_entry_fn_t)primary_heartbeat_fn, c);
return 0;
}
void ha_start_primary_heartbeat(cluster_t *c)
{
if (!c) return;
ha_state_t *ha = &c->ha;
int interval_ms = (int)ha->config.heartbeat_interval_ms;
if (interval_ms <= 0)
interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS;
/* Guard: skip if delaymgr is not running (e.g. unit tests with zeroed cluster_t).
* In production the delaymgr is always started before any promotion. */
if (!c->delaymgr.running)
return;
/* One-shot: the task re-arms itself each time while role == PRIMARY */
int rc = delay_task_ms(&c->delaymgr, interval_ms, 1,
(delay_task_entry_fn_t)primary_heartbeat_fn, c);
if (rc < 0)
WARN("ha_start_primary_heartbeat: delay_task_ms failed: %d", rc);
else
INFO("HA: primary heartbeat armed (interval=%dms)", interval_ms);
}