forked from lei.chen/myxce
Rongsong Shen
d00d3bb051
- ha.c: add raft_cb_trace callback wired to s_real_cbs.on_trace; emits RAFT[pN] term=T event=E at DEBUG3 level, visible with -vvv flag. - ha.c: repl_ack_decode_and_dispatch now silently drops OPCode_nop and OPCode_reply instead of WARNING — heartbeat NOP replies from standbys arrive on promoted endpoints and are expected. - deploy-ha.sh: fix --vcpu -> --vcpus; add \$VERBOSE passthrough so start.sh respects VERBOSE=-vvv from the environment.
3071 lines
114 KiB
C
3071 lines
114 KiB
C
#define _GNU_SOURCE
|
||
#include "ha.h"
|
||
|
||
#include <arpa/inet.h>
|
||
#include <errno.h>
|
||
#include <fcntl.h>
|
||
#include <inttypes.h>
|
||
#include <netdb.h>
|
||
#include <poll.h>
|
||
#include <pthread.h>
|
||
#include <signal.h>
|
||
#include <stdatomic.h>
|
||
#include <stdio.h>
|
||
#include <stdlib.h>
|
||
#include <string.h>
|
||
#include <sys/epoll.h>
|
||
#include <sys/socket.h>
|
||
#include <time.h>
|
||
#include <unistd.h>
|
||
#include <zlib.h>
|
||
|
||
#include <openssl/ssl.h>
|
||
#include <openssl/err.h>
|
||
|
||
/* Circular-include safety: cluster_def.h includes ha.h (include-guarded),
|
||
* and ha.h only forward-declares cluster_t — no cycle. */
|
||
#include "xce/cluster_def.h"
|
||
#include "xce/errnox.h"
|
||
#include "controller.capnp.h"
|
||
#include "persist.h"
|
||
#include "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);
|
||
|
||
/* ── 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 with lower config_tag */
|
||
#ifdef XCE_TESTING
|
||
if (ha_push_config_hook != NULL)
|
||
ha_push_config_hook(c);
|
||
#else
|
||
push_config_to_standbys(c, NULL, 0, NULL, 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
|
||
/* Arm committing timeout watchdog */
|
||
committing_timeout_ctx_t *ct = calloc(1, sizeof(*ct));
|
||
if (ct) {
|
||
atomic_init(&ct->cancelled, 0);
|
||
ct->ha = ha;
|
||
ct->c = c;
|
||
int rc = delay_task_ms(&c->delaymgr,
|
||
(int)ha->config.committing_timeout_ms,
|
||
1, committing_timeout_fn, ct);
|
||
if (rc < 0) {
|
||
WARN("raft_cb_become_leader: delay_task_ms for committing timeout failed: %d",
|
||
rc);
|
||
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);
|
||
|
||
/* 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);
|
||
atomic_store_explicit(&ha->failover_requested, 0, memory_order_relaxed);
|
||
atomic_store_explicit(&ha->watchdog_stop, 0, memory_order_relaxed);
|
||
|
||
pthread_mutex_init(&ha->lock, NULL);
|
||
|
||
ha->ssl_ctx = NULL;
|
||
ha->jitter_seed = 0;
|
||
ha->standby_endpoints = NULL;
|
||
ha->primary_ctrl_fd = -1;
|
||
|
||
/* election-iomgr-refactor Task 2.1: explicit zero-init for documentation clarity
|
||
* (memset above already zeroes these, but being explicit is better documentation) */
|
||
ha->client_ssl_ctx = NULL;
|
||
ha->phase2_done = 0;
|
||
ha->n_peers = 0;
|
||
atomic_store_explicit(&ha->repl_ack_count, 0, memory_order_relaxed);
|
||
|
||
/* heartbeat_seen starts at 0: no heartbeat seen yet.
|
||
* ha_wait_for_heartbeat_or_timeout() calls sem_timedwait on this. */
|
||
sem_init(&ha->heartbeat_seen, 0, 0);
|
||
|
||
/* ── 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;
|
||
}
|
||
|
||
/* ── Promotion / demotion / SIGUSR1 ────────────────────────── */
|
||
|
||
/* Async-signal-safe: atomic store on int-width type (POSIX.1-2017 §2.4.3).
|
||
* Pointer set once before signals are enabled — no locking required. */
|
||
static ha_state_t *g_sigusr1_ha = NULL;
|
||
|
||
static void sigusr1_handler(int sig)
|
||
{
|
||
(void)sig;
|
||
if (g_sigusr1_ha)
|
||
atomic_store_explicit(&g_sigusr1_ha->failover_requested, 1,
|
||
memory_order_release);
|
||
}
|
||
|
||
void ha_install_sigusr1(ha_state_t *ha)
|
||
{
|
||
g_sigusr1_ha = ha;
|
||
struct sigaction sa;
|
||
memset(&sa, 0, sizeof(sa));
|
||
sa.sa_handler = sigusr1_handler;
|
||
sigemptyset(&sa.sa_mask);
|
||
sa.sa_flags = SA_RESTART;
|
||
sigaction(SIGUSR1, &sa, NULL);
|
||
}
|
||
|
||
void ha_do_promote(ha_state_t *ha)
|
||
{
|
||
/* Monotonically increasing fencing token — standbys reject entries from lower terms */
|
||
atomic_fetch_add_explicit(&ha->term, 1, memory_order_acq_rel);
|
||
atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY, memory_order_release);
|
||
/* COH-5: new primary starts a fresh CRC chain; standbys reset their chain on catch-up */
|
||
ha->last_received_crc = WAL_CRC_NONE;
|
||
/* Req 10.5: log FSM transition to PRIMARY */
|
||
INFO("HA FSM: → CTRL_PRIMARY (term=%" PRIu64 ")",
|
||
atomic_load_explicit(&ha->term, memory_order_acquire));
|
||
}
|
||
|
||
void ha_do_demote(ha_state_t *ha)
|
||
{
|
||
atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY, memory_order_release);
|
||
/* Reset so the next ha_promote_to_primary() re-registers hooks idempotently */
|
||
ha->hooks_registered = 0;
|
||
/* Hook disable requires hooksys_t — done by ha_demote_to_standby(cluster_t*) */
|
||
/* Req 10.5: log FSM transition to STANDBY */
|
||
INFO("HA FSM: → CTRL_STANDBY (term=%" PRIu64 ")",
|
||
atomic_load_explicit(&ha->term, memory_order_acquire));
|
||
}
|
||
|
||
/* ── P1-6: Election probing ─────────────────────────────────── */
|
||
|
||
/*
|
||
* probe_thread: attempt a TCP (optionally SSL) connection to one candidate.
|
||
* Sets ctx->got_ack = 1 if the connection succeeds before election_done is set.
|
||
*/
|
||
static void *probe_thread(void *arg)
|
||
{
|
||
probe_ctx_t *ctx = (probe_ctx_t *)arg;
|
||
|
||
struct addrinfo hints, *res = NULL;
|
||
memset(&hints, 0, sizeof(hints));
|
||
hints.ai_family = AF_UNSPEC;
|
||
hints.ai_socktype = SOCK_STREAM;
|
||
|
||
char port_str[16];
|
||
snprintf(port_str, sizeof(port_str), "%u", (unsigned)ctx->candidate->port);
|
||
if (getaddrinfo(ctx->candidate->hostname, port_str, &hints, &res) != 0)
|
||
goto done;
|
||
|
||
int sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
|
||
if (sock < 0) { freeaddrinfo(res); goto done; }
|
||
|
||
/* Non-blocking connect */
|
||
int flags = fcntl(sock, F_GETFL, 0);
|
||
fcntl(sock, F_SETFL, flags | O_NONBLOCK);
|
||
connect(sock, res->ai_addr, res->ai_addrlen);
|
||
freeaddrinfo(res);
|
||
res = NULL;
|
||
|
||
/* Poll loop: check election_done between iterations */
|
||
while (!atomic_load(ctx->election_done)) {
|
||
struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP };
|
||
int pr = poll(&pfd, 1, (int)ctx->timeout_ms);
|
||
if (pr < 0 && errno == EINTR) continue;
|
||
if (pr == 0) goto cleanup; /* poll timeout */
|
||
if (pfd.revents & (POLLERR | POLLHUP)) goto cleanup;
|
||
if (pfd.revents & POLLOUT) break;
|
||
}
|
||
if (atomic_load(ctx->election_done)) goto cleanup;
|
||
|
||
/* Verify connect succeeded */
|
||
{
|
||
int err = 0;
|
||
socklen_t el = sizeof(err);
|
||
getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
|
||
if (err != 0) goto cleanup;
|
||
}
|
||
|
||
/* SSL handshake — only if ssl_ctx is set */
|
||
if (ctx->ssl_ctx) {
|
||
SSL *ssl = SSL_new((SSL_CTX *)ctx->ssl_ctx);
|
||
if (!ssl) goto cleanup;
|
||
/* Restore blocking for SSL */
|
||
fcntl(sock, F_SETFL, flags & ~O_NONBLOCK);
|
||
SSL_set_fd(ssl, sock);
|
||
int sr;
|
||
do {
|
||
sr = SSL_connect(ssl);
|
||
if (sr <= 0) {
|
||
int ssl_err = SSL_get_error(ssl, sr);
|
||
if (ssl_err != SSL_ERROR_WANT_READ &&
|
||
ssl_err != SSL_ERROR_WANT_WRITE)
|
||
break;
|
||
}
|
||
} while (sr <= 0 && !atomic_load(ctx->election_done));
|
||
SSL_shutdown(ssl);
|
||
SSL_free(ssl);
|
||
if (sr <= 0 || atomic_load(ctx->election_done))
|
||
goto cleanup;
|
||
}
|
||
|
||
ctx->got_ack = 1;
|
||
|
||
cleanup:
|
||
close(sock);
|
||
done:
|
||
return NULL;
|
||
}
|
||
|
||
/*
|
||
* wait_any_probe: spin-poll until any probe sets got_ack, or deadline passes.
|
||
* Returns 1 if any probe acknowledged, 0 if deadline expired.
|
||
*/
|
||
static int wait_any_probe(probe_ctx_t *probes, int n,
|
||
const struct timespec *deadline)
|
||
{
|
||
while (1) {
|
||
for (int i = 0; i < n; i++) {
|
||
if (probes[i].got_ack)
|
||
return 1;
|
||
}
|
||
struct timespec now;
|
||
clock_gettime(CLOCK_MONOTONIC, &now);
|
||
if (now.tv_sec > deadline->tv_sec ||
|
||
(now.tv_sec == deadline->tv_sec &&
|
||
now.tv_nsec >= deadline->tv_nsec))
|
||
return 0;
|
||
/* Sleep 10ms between checks */
|
||
struct timespec sleep_ts = { .tv_sec = 0, .tv_nsec = 10 * 1000000L };
|
||
nanosleep(&sleep_ts, NULL);
|
||
}
|
||
}
|
||
|
||
/*
|
||
* am_i_primary: election probe logic.
|
||
*
|
||
* Priority-0 is unconditionally primary — never probes.
|
||
* Priority-N probes the N candidates with lower index (higher priority).
|
||
* If any responds before election_timeout + priority_step*N + jitter, this
|
||
* node becomes standby. If deadline expires with no response, self-promote.
|
||
*/
|
||
int am_i_primary(cluster_t *c)
|
||
{
|
||
/* Priority-0 never probes — it is unconditionally primary */
|
||
if (c->ha.my_priority == 0)
|
||
return 1;
|
||
|
||
const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
|
||
int n_probes = (c->ha.my_priority < pcfg->candidates_count)
|
||
? c->ha.my_priority
|
||
: pcfg->candidates_count;
|
||
if (n_probes == 0)
|
||
return 1;
|
||
|
||
/* Seed jitter PRNG if needed */
|
||
if (c->ha.jitter_seed == 0)
|
||
ha_seed_jitter(&c->ha.jitter_seed, c->ha.my_priority);
|
||
|
||
atomic_int election_done;
|
||
atomic_init(&election_done, 0);
|
||
probe_ctx_t probes[MAX_HA_CANDIDATES];
|
||
|
||
for (int i = 0; i < n_probes; i++) {
|
||
probes[i].candidate = (xce_candidate_t *)&pcfg->candidates[i];
|
||
probes[i].ssl_ctx = c->ha.ssl_ctx;
|
||
probes[i].timeout_ms = c->ha.config.probe_connect_timeout_ms;
|
||
probes[i].election_done = &election_done;
|
||
probes[i].got_ack = 0;
|
||
pthread_create(&probes[i].thread, NULL, probe_thread, &probes[i]);
|
||
}
|
||
|
||
/* Compute deadline = election_timeout + priority_step * my_priority + jitter */
|
||
long to_ms = (long)c->ha.config.election_timeout_ms
|
||
+ (long)c->ha.my_priority
|
||
* (long)c->ha.config.election_priority_step_ms;
|
||
if (c->ha.config.election_jitter_max_ms > 0) {
|
||
to_ms += (long)(xorshift64_next(&c->ha.jitter_seed)
|
||
% ((uint64_t)c->ha.config.election_jitter_max_ms + 1));
|
||
}
|
||
|
||
struct timespec deadline;
|
||
clock_gettime(CLOCK_MONOTONIC, &deadline);
|
||
deadline.tv_sec += to_ms / 1000;
|
||
deadline.tv_nsec += (to_ms % 1000) * 1000000L;
|
||
if (deadline.tv_nsec >= 1000000000L) {
|
||
deadline.tv_sec++;
|
||
deadline.tv_nsec -= 1000000000L;
|
||
}
|
||
|
||
int got_ack = wait_any_probe(probes, n_probes, &deadline);
|
||
|
||
/* Signal all threads to stop */
|
||
atomic_store(&election_done, 1);
|
||
for (int i = 0; i < n_probes; i++)
|
||
pthread_join(probes[i].thread, NULL);
|
||
|
||
if (got_ack) {
|
||
WARN("HA election: higher-priority candidate responded — "
|
||
"starting as standby (my_priority=%d)", c->ha.my_priority);
|
||
return 0;
|
||
}
|
||
INFO("HA election: no higher-priority candidate responded within %ldms — "
|
||
"starting as primary (my_priority=%d)", to_ms, c->ha.my_priority);
|
||
return 1;
|
||
}
|
||
|
||
/*
|
||
* ha_is_primary: lock-free role check.
|
||
*/
|
||
int ha_is_primary(const ha_state_t *ha)
|
||
{
|
||
return (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire)
|
||
== CTRL_PRIMARY;
|
||
}
|
||
|
||
/* ha_check_priority_stepdown() deleted — replaced by quorum election (Req 2.1) */
|
||
|
||
/*
|
||
* start_heartbeat_listener / stop_heartbeat_listener:
|
||
* Stub implementations — full listener is implemented in Task 7 (cluster.c).
|
||
* The stubs allow am_i_primary() callers (main.c / cluster.c) to compile and
|
||
* link without requiring the full Phase-2 network stack.
|
||
*/
|
||
void start_heartbeat_listener(cluster_t *c)
|
||
{
|
||
(void)c;
|
||
/* Heartbeat listener stub — full implementation in cluster.c Task 7 */
|
||
}
|
||
|
||
void stop_heartbeat_listener(cluster_t *c)
|
||
{
|
||
(void)c;
|
||
}
|
||
|
||
/* ── P3-6: Test-only failover shim ────────────────────────────── */
|
||
|
||
#ifdef XCE_TESTING
|
||
uint64_t ha_trigger_manual_failover_for_test(ha_state_t *ha)
|
||
{
|
||
ha_do_demote(ha);
|
||
ha_do_promote(ha);
|
||
return atomic_load_explicit(&ha->term, memory_order_acquire);
|
||
}
|
||
#endif
|
||
|
||
/* ha_check_heartbeat_nocluster() removed from the public API — logic inlined in
|
||
* ha_watchdog_thread(). XCE_TESTING shim preserved so existing unit tests compile. */
|
||
#ifdef XCE_TESTING
|
||
void ha_check_heartbeat_nocluster(ha_state_t *ha)
|
||
{
|
||
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
|
||
if (role != CTRL_STANDBY) return;
|
||
if (ha->primary_host[0] == '\0') return;
|
||
|
||
struct timespec now;
|
||
clock_gettime(CLOCK_MONOTONIC, &now);
|
||
pthread_mutex_lock(&ha->lock);
|
||
struct timespec last = ha->last_heartbeat;
|
||
pthread_mutex_unlock(&ha->lock);
|
||
|
||
uint64_t elapsed_ms =
|
||
(uint64_t)(now.tv_sec - last.tv_sec) * 1000ULL +
|
||
(uint64_t)(now.tv_nsec - last.tv_nsec) / 1000000ULL;
|
||
|
||
if (elapsed_ms >= ha->config.heartbeat_timeout_ms)
|
||
atomic_store_explicit(&ha->failover_requested, 1, memory_order_release);
|
||
}
|
||
#endif /* XCE_TESTING */
|
||
|
||
static void *ha_watchdog_thread(void *arg)
|
||
{
|
||
ha_state_t *ha = arg;
|
||
|
||
while (!atomic_load_explicit(&ha->watchdog_stop, memory_order_acquire)) {
|
||
long ms = (long)ha->config.heartbeat_interval_ms;
|
||
struct timespec ts = {
|
||
.tv_sec = ms / 1000,
|
||
.tv_nsec = (ms % 1000) * 1000000L,
|
||
};
|
||
nanosleep(&ts, NULL);
|
||
|
||
/* Re-check stop flag after sleep */
|
||
if (atomic_load_explicit(&ha->watchdog_stop, memory_order_acquire)) break;
|
||
|
||
/* Heartbeat timeout check (inlined from deleted ha_check_heartbeat_nocluster) */
|
||
{
|
||
ctrl_role_t role =
|
||
(ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
|
||
if (role == CTRL_STANDBY && ha->primary_host[0] != '\0') {
|
||
struct timespec now;
|
||
clock_gettime(CLOCK_MONOTONIC, &now);
|
||
pthread_mutex_lock(&ha->lock);
|
||
struct timespec last = ha->last_heartbeat;
|
||
pthread_mutex_unlock(&ha->lock);
|
||
uint64_t elapsed_ms =
|
||
(uint64_t)(now.tv_sec - last.tv_sec) * 1000ULL +
|
||
(uint64_t)(now.tv_nsec - last.tv_nsec) / 1000000ULL;
|
||
if (elapsed_ms >= ha->config.heartbeat_timeout_ms)
|
||
atomic_store_explicit(&ha->failover_requested, 1,
|
||
memory_order_release);
|
||
}
|
||
}
|
||
|
||
if (atomic_load_explicit(&ha->failover_requested, memory_order_acquire)) {
|
||
atomic_store_explicit(&ha->failover_requested, 0, memory_order_release);
|
||
/* Promotion requires cluster_t; the watchdog only has ha_state_t.
|
||
* Full ha_promote_to_primary(c) is called from the cluster's own
|
||
* failover path which holds cluster_t context. */
|
||
}
|
||
}
|
||
return NULL;
|
||
}
|
||
|
||
int ha_start_watchdog(ha_state_t *ha)
|
||
{
|
||
atomic_store_explicit(&ha->watchdog_stop, 0, memory_order_release);
|
||
int rc = pthread_create(&ha->watchdog_thread, NULL, ha_watchdog_thread, ha);
|
||
if (rc != 0) return -rc;
|
||
return 0;
|
||
}
|
||
|
||
void ha_stop_watchdog(ha_state_t *ha)
|
||
{
|
||
atomic_store_explicit(&ha->watchdog_stop, 1, memory_order_release);
|
||
if (ha->watchdog_thread != 0) {
|
||
pthread_join(ha->watchdog_thread, NULL);
|
||
ha->watchdog_thread = 0;
|
||
}
|
||
}
|
||
|
||
void ha_fini(ha_state_t *ha)
|
||
{
|
||
if (!ha)
|
||
return;
|
||
ha_stop_watchdog(ha); /* join watchdog thread before destroying mutex */
|
||
pthread_mutex_destroy(&ha->lock);
|
||
sem_destroy(&ha->heartbeat_seen);
|
||
|
||
/* Fix ASAN leak: free the SSL context if one was allocated */
|
||
if (ha->ssl_ctx) {
|
||
SSL_CTX_free((SSL_CTX *)ha->ssl_ctx);
|
||
ha->ssl_ctx = NULL;
|
||
}
|
||
}
|
||
|
||
/* ── Dispatch gate (req 1.3, 1.4) ──────────────────────────────────────────
|
||
*
|
||
* Called from myreq_handler() in cluster.c before the opcode switch.
|
||
* Returns 0 when the operation is permitted for the current role.
|
||
*
|
||
* Primary-only ops are any op that is not one of the always-allowed set:
|
||
* nop, reply, hello, join, shutdown, heartbeat, redirect,
|
||
* steppingDown, replEntry, replAck, replSnapshot, replCatchUpDone,
|
||
* replConfig, replConfigPush, requestVote, probeConfigTag,
|
||
* walAck, upgradeConfigAck, snapshotTransfer, walCatchupEntry.
|
||
*
|
||
* For CTRL_STANDBY the check is: op is in the primary-only set → deny.
|
||
* For CTRL_INIT / CTRL_ELECTING the allowlist is tighter: only
|
||
* OP_requestVote and OP_probeConfigTag pass through.
|
||
*/
|
||
int ha_dispatch_gate(ha_state_t *ha, int opcode)
|
||
{
|
||
ctrl_role_t role =
|
||
(ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
|
||
|
||
switch (role) {
|
||
case CTRL_PRIMARY:
|
||
return 0;
|
||
|
||
case CTRL_INIT:
|
||
/* Allow election probes, config probes, AND heartbeats so the primary's
|
||
* periodic heartbeat can reach the handler and post heartbeat_seen,
|
||
* triggering the CTRL_INIT → CTRL_STANDBY transition (Req 1.1). */
|
||
if (opcode == OPCode_requestVote ||
|
||
opcode == OPCode_probeConfigTag ||
|
||
opcode == OPCode_heartbeat)
|
||
return 0;
|
||
return -XCE_ERR_NOT_READY;
|
||
|
||
case CTRL_ELECTING:
|
||
if (opcode == OPCode_requestVote || opcode == OPCode_probeConfigTag)
|
||
return 0;
|
||
return -XCE_ERR_NOT_READY;
|
||
|
||
case CTRL_PRIMARY_COMMITTING:
|
||
/* Req 5.1, 5.2, 8.1: Primary won election but has not yet received
|
||
* majority replication ack. Allow only replication-related and
|
||
* administrative opcodes; deny job dispatch and queue management
|
||
* with XCE_ERR_NOT_READY. */
|
||
switch (opcode) {
|
||
case OPCode_heartbeat:
|
||
case OPCode_replConfig:
|
||
case OPCode_replConfigPush:
|
||
case OPCode_upgradeConfigAck:
|
||
case OPCode_hello:
|
||
case OPCode_walAck:
|
||
case OPCode_replEntry:
|
||
case OPCode_replAck:
|
||
case OPCode_replCatchUpDone:
|
||
return 0;
|
||
default:
|
||
return -XCE_ERR_NOT_READY;
|
||
}
|
||
|
||
case CTRL_STANDBY: {
|
||
/* Allow ops that standbys legitimately handle */
|
||
switch (opcode) {
|
||
case OPCode_nop:
|
||
case OPCode_reply:
|
||
case OPCode_hello:
|
||
case OPCode_join:
|
||
case OPCode_shutdown:
|
||
case OPCode_heartbeat:
|
||
case OPCode_redirect:
|
||
case OPCode_steppingDown:
|
||
case OPCode_replEntry:
|
||
case OPCode_replAck:
|
||
case OPCode_replSnapshot:
|
||
case OPCode_replCatchUpDone:
|
||
case OPCode_replConfig:
|
||
case OPCode_replConfigPush:
|
||
case OPCode_requestVote:
|
||
case OPCode_probeConfigTag:
|
||
case OPCode_walAck:
|
||
case OPCode_upgradeConfigAck:
|
||
case OPCode_snapshotTransfer:
|
||
case OPCode_walCatchupEntry:
|
||
return 0;
|
||
default:
|
||
return -XCE_ERR_STANDBY;
|
||
}
|
||
}
|
||
|
||
default:
|
||
return 0;
|
||
}
|
||
}
|
||
|
||
/* ── Redirect helpers used by join.c ───────────────────────── */
|
||
|
||
/*
|
||
* ha_should_redirect: returns 1 when this standby knows the primary's
|
||
* address and should redirect incoming join requests there.
|
||
* Returns 0 for PRIMARY role, or for STANDBY with no known primary yet.
|
||
*/
|
||
int ha_should_redirect(const ha_state_t *ha)
|
||
{
|
||
ctrl_role_t role =
|
||
(ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
|
||
if (role == CTRL_PRIMARY)
|
||
return 0;
|
||
if (ha->primary_host[0] == '\0' || ha->primary_port == 0)
|
||
return 0;
|
||
return 1;
|
||
}
|
||
|
||
/*
|
||
* ha_no_primary_known: returns 1 when this node is a STANDBY but does not
|
||
* yet know the primary's address. Callers should reply E_XCE_NOMASTER so
|
||
* the client retries rather than treating the error as permanent.
|
||
*/
|
||
int ha_no_primary_known(const ha_state_t *ha)
|
||
{
|
||
ctrl_role_t role =
|
||
(ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
|
||
if (role == CTRL_PRIMARY)
|
||
return 0;
|
||
return (ha->primary_host[0] == '\0' || ha->primary_port == 0) ? 1 : 0;
|
||
}
|
||
|
||
/*
|
||
* ha_promote_to_primary: increment term, set role=PRIMARY, then register
|
||
* persistence hooks (idempotency guard via hooks_registered).
|
||
* Safe to call on a zero-init cluster_t (zeroed hooksys → xce_add_hook
|
||
* returns -EINVAL harmlessly; hooks_registered is still set to 1).
|
||
*/
|
||
int ha_promote_to_primary(cluster_t *c)
|
||
{
|
||
ha_do_promote(&c->ha);
|
||
|
||
/* Register persistence hooks — idempotency guard prevents double-add */
|
||
if (c->ha.hooks_registered == 0) {
|
||
xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_JOB,
|
||
(hook_fn_t)persist_job_hook_fun, &c->persist);
|
||
xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_QUEUE,
|
||
(hook_fn_t)persist_queue_hook_fun, &c->persist);
|
||
xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER,
|
||
(hook_fn_t)persist_worker_hook_fun, &c->persist);
|
||
c->ha.hooks_registered = 1;
|
||
}
|
||
|
||
/* 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 6.1, 6.2: role-based deny gate — checked before any Raft state mutation.
|
||
* PRIMARY and PRIMARY_COMMITTING never grant votes (protect the live primary).
|
||
* STANDBY with a fresh heartbeat also denies (primary is alive, no new election
|
||
* needed). CTRL_INIT and CTRL_ELECTING fall through to Raft evaluation. */
|
||
ctrl_role_t my_role = (ctrl_role_t)atomic_load_explicit(&ha->role,
|
||
memory_order_acquire);
|
||
if (my_role == CTRL_PRIMARY || my_role == CTRL_PRIMARY_COMMITTING)
|
||
return 0; /* deny — I am primary (or committing); protect against challengers */
|
||
if (my_role == CTRL_STANDBY && ha_heartbeat_is_fresh(ha))
|
||
return 0; /* deny — my primary is alive */
|
||
|
||
/* Req 2.5: catching-up standbys deny all votes */
|
||
if (in_catch_up_flag)
|
||
return 0;
|
||
|
||
/* Protect voted_for check+set from concurrent vote requests on different vtasks */
|
||
pthread_mutex_lock(&ha->lock);
|
||
|
||
uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire);
|
||
|
||
/* Req 2.2: stale term → deny, do not update any state */
|
||
if (term < my_term) {
|
||
pthread_mutex_unlock(&ha->lock);
|
||
return 0;
|
||
}
|
||
|
||
/* Req 2.3: higher term → advance term, clear prior vote record */
|
||
if (term > my_term) {
|
||
atomic_store_explicit(&ha->term, term, memory_order_release);
|
||
ha->voted_for_term = 0;
|
||
memset(ha->voted_for_name, 0, sizeof(ha->voted_for_name));
|
||
}
|
||
|
||
/* Req 2.6: already voted for a different candidate in this term → deny */
|
||
if (ha->voted_for_term == term &&
|
||
ha->voted_for_name[0] != '\0' &&
|
||
strcmp(ha->voted_for_name, candidate_id) != 0) {
|
||
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;
|
||
|
||
#endif /* XCE_TESTING */
|
||
|
||
/* ── Election Evaluation (Req 2.1, 2.7, 2.8) ───────────────────────────────
|
||
*
|
||
* ha_elect_evaluate_and_signal: evaluate accumulated votes; set ctx->result
|
||
* and post ctx->done when quorum is reached or all peers replied.
|
||
*
|
||
* Quorum = ceil((n_peers + 1) / 2) where n_peers+1 is total cluster size.
|
||
* ctx->votes_granted starts at 1 (self-vote); ctx->n_peers is n_other_peers.
|
||
* Quorum = ceil((n_peers + 1) / 2) = (n_peers + 1 + 1) / 2 = (n_peers + 2) / 2.
|
||
*
|
||
* Yield gate (Req 2.7): if any responding peer has a higher config_tag, or
|
||
* equal config_tag with a higher LSN than the candidate → yield (CTRL_STANDBY).
|
||
* Candidate's own tag/lsn is stored in ctx->votes[0].my_tag / my_lsn.
|
||
*
|
||
* Fatal gate (Req 2.8): if candidate has config_tag=0 and any peer has tag>0
|
||
* → CTRL_STANDBY (never PRIMARY with no config when peer has config).
|
||
*
|
||
* Returns CTRL_PRIMARY or CTRL_STANDBY; also sets ctx->result.
|
||
*/
|
||
ctrl_role_t ha_elect_evaluate_and_signal(ha_state_t *ha, ha_elect_ctx_t *ctx)
|
||
{
|
||
/* Guard: called from handle_request_vote_reply under ctx->lock; called from
|
||
* unit-test path (peers==NULL) without threads — done_posted prevents double
|
||
* execution if two threads both hit quorum simultaneously. */
|
||
if (ctx->done_posted)
|
||
return (ctx->result == 1) ? CTRL_PRIMARY : CTRL_STANDBY;
|
||
|
||
ctrl_role_t role_result;
|
||
|
||
/* Fix 2 (CRITICAL): zero-peer guard — single-node shortcircuit.
|
||
* n_peers==0 means no other nodes; votes[] flexible array has 0 entries,
|
||
* so accessing votes[0] would be out-of-bounds. Self-vote (votes_granted=1)
|
||
* always satisfies quorum=1. No fresher-peer or fatal-gate checks apply. */
|
||
if (ctx->n_peers == 0) {
|
||
ctx->result = 1;
|
||
ctx->done_posted = 1;
|
||
sem_post(&ctx->done);
|
||
role_result = CTRL_PRIMARY;
|
||
goto apply_role;
|
||
}
|
||
|
||
{
|
||
int total_nodes = ctx->n_peers + 1; /* include self */
|
||
int quorum = (total_nodes + 1) / 2; /* ceil(total/2) */
|
||
|
||
/* Req 2.8: fatal gate — candidate has no config but a peer does */
|
||
uint64_t my_tag = ctx->votes[0].my_tag;
|
||
uint64_t my_lsn = ctx->votes[0].my_lsn;
|
||
if (my_tag == 0 && ctx->max_resp_tag > 0) {
|
||
ctx->result = 0;
|
||
ctx->done_posted = 1;
|
||
sem_post(&ctx->done);
|
||
role_result = CTRL_STANDBY;
|
||
goto apply_role;
|
||
}
|
||
|
||
/* Check quorum */
|
||
if (ctx->votes_granted < quorum) {
|
||
ctx->result = 0;
|
||
ctx->done_posted = 1;
|
||
sem_post(&ctx->done);
|
||
role_result = CTRL_STANDBY;
|
||
goto apply_role;
|
||
}
|
||
|
||
/* Req 2.7: yield gate — fresher responding peer */
|
||
if (ctx->max_resp_tag > my_tag) {
|
||
ctx->result = 0;
|
||
ctx->done_posted = 1;
|
||
sem_post(&ctx->done);
|
||
role_result = CTRL_STANDBY;
|
||
goto apply_role;
|
||
}
|
||
if (ctx->max_resp_tag == my_tag && ctx->max_resp_lsn > my_lsn) {
|
||
ctx->result = 0;
|
||
ctx->done_posted = 1;
|
||
sem_post(&ctx->done);
|
||
role_result = CTRL_STANDBY;
|
||
goto apply_role;
|
||
}
|
||
|
||
/* Quorum met and no fresher peer → PRIMARY */
|
||
ctx->result = 1;
|
||
ctx->done_posted = 1;
|
||
sem_post(&ctx->done);
|
||
role_result = CTRL_PRIMARY;
|
||
}
|
||
|
||
apply_role:
|
||
/* ── Task 4.1: apply role and call cluster_init_phase2 ─────────────────
|
||
*
|
||
* After determining the election result, update ha->role and (on first boot)
|
||
* call cluster_init_phase2. This runs once per process lifetime — the
|
||
* phase2_done flag prevents re-entry on subsequent elections (RDHUP failover).
|
||
*
|
||
* The cluster_t pointer is carried in ctx->c (NULL in unit tests).
|
||
*
|
||
* Cancel the election timeout watchdog if we won (it would be a no-op now
|
||
* but the explicit cancel avoids a spurious second evaluate call).
|
||
*/
|
||
{
|
||
cluster_t *c = ctx->c;
|
||
int is_first_boot = !ha->phase2_done;
|
||
|
||
if (role_result == CTRL_PRIMARY) {
|
||
/* Cancel election timeout watchdog (quorum already reached) */
|
||
if (ctx->timeout_ctx) {
|
||
election_timeout_ctx_t *etc =
|
||
(election_timeout_ctx_t *)ctx->timeout_ctx;
|
||
atomic_store(&etc->cancelled, 1);
|
||
}
|
||
|
||
/* First boot: call cluster_init_phase2 to load config+data (Req 9.2) */
|
||
if (is_first_boot) {
|
||
#ifdef XCE_TESTING
|
||
/* In test mode, use hook if set; skip real phase2 (cluster.c not linked) */
|
||
if (ha_phase2_hook != NULL)
|
||
ha_phase2_hook(c);
|
||
#else
|
||
if (c != NULL)
|
||
cluster_init_phase2(c);
|
||
#endif
|
||
ha->phase2_done = 1;
|
||
}
|
||
|
||
/* Enter COMMITTING: block job traffic until majority ack (Req 4.1, 8.1) */
|
||
atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY_COMMITTING,
|
||
memory_order_release);
|
||
INFO("HA FSM: → CTRL_PRIMARY_COMMITTING (term=%" PRIu64
|
||
", first_boot=%d)", atomic_load_explicit(&ha->term, memory_order_acquire),
|
||
is_first_boot);
|
||
|
||
/* Reset ack counter for this committing round (Req 5.2) */
|
||
atomic_store(&ha->repl_ack_count, 0);
|
||
|
||
/* Push config to standbys with lower config_tag (Req 7.1) */
|
||
#ifdef XCE_TESTING
|
||
if (ha_push_config_hook != NULL)
|
||
ha_push_config_hook(c);
|
||
#else
|
||
if (c != NULL)
|
||
push_config_to_standbys(c, NULL, 0, NULL, 0);
|
||
#endif
|
||
|
||
/* Send CtrlHelloRequest to all standbys to trigger WAL catch-up (Req 7.2) */
|
||
#ifdef XCE_TESTING
|
||
if (ha_send_ctrlhello_hook != NULL)
|
||
ha_send_ctrlhello_hook(c);
|
||
#else
|
||
if (c != NULL)
|
||
ha_send_ctrlhello_to_all_standbys(c);
|
||
#endif
|
||
|
||
/* Arm committing timeout watchdog (degraded-mode fallback, Req 8.4) */
|
||
if (c) {
|
||
committing_timeout_ctx_t *ct = calloc(1, sizeof(*ct));
|
||
if (ct) {
|
||
atomic_init(&ct->cancelled, 0);
|
||
ct->ha = ha;
|
||
ct->c = c;
|
||
int rc = delay_task_ms(&c->delaymgr,
|
||
(int)ha->config.committing_timeout_ms,
|
||
1, committing_timeout_fn, ct);
|
||
if (rc < 0) {
|
||
WARN("ha_elect_evaluate_and_signal: delay_task_ms "
|
||
"for committing timeout failed: %d", rc);
|
||
free(ct);
|
||
}
|
||
}
|
||
}
|
||
} else {
|
||
/* STANDBY result */
|
||
|
||
/* Cancel election timeout watchdog — it is now a no-op but explicit
|
||
* cancellation matches spec wording and avoids a spurious lock-acquire
|
||
* if the timeout fires after STANDBY is decided. */
|
||
if (ctx->timeout_ctx) {
|
||
election_timeout_ctx_t *etc =
|
||
(election_timeout_ctx_t *)ctx->timeout_ctx;
|
||
atomic_store(&etc->cancelled, 1);
|
||
}
|
||
|
||
/* First boot: call cluster_init_phase2 so structures are ready;
|
||
* config arrives via primary's replication push (Req 9.2, 7.3) */
|
||
if (is_first_boot) {
|
||
#ifdef XCE_TESTING
|
||
/* In test mode, use hook if set; skip real phase2 (cluster.c not linked) */
|
||
if (ha_phase2_hook != NULL)
|
||
ha_phase2_hook(c);
|
||
#else
|
||
if (c != NULL)
|
||
cluster_init_phase2(c);
|
||
#endif
|
||
ha->phase2_done = 1;
|
||
}
|
||
|
||
atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY,
|
||
memory_order_release);
|
||
INFO("HA FSM: → CTRL_STANDBY (term=%" PRIu64 ", first_boot=%d)",
|
||
atomic_load_explicit(&ha->term, memory_order_acquire),
|
||
is_first_boot);
|
||
/* 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.
|
||
* Expects OPCode_replCatchUpDone; calls ha_repl_ack_received on success.
|
||
* Drops unexpected opcodes with a WARN log.
|
||
*/
|
||
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;
|
||
capn_free(&capn_ctx);
|
||
|
||
if (opcode == OPCode_replCatchUpDone) {
|
||
ha_repl_ack_received(ha, c);
|
||
} else if (opcode == OPCode_nop || opcode == OPCode_reply) {
|
||
/* Heartbeat NOP reply or generic reply — silently drop */
|
||
} else {
|
||
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);
|
||
}
|
||
|
||
/*
|
||
* 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;
|
||
|
||
/* 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);
|
||
}
|
||
|
||
/* ── Heartbeat timeout watchdog ─────────────────────────────────────────── */
|
||
|
||
typedef struct {
|
||
ha_state_t *ha;
|
||
cluster_t *c;
|
||
sem_t reset_sem; /* ha_on_heartbeat_received posts here */
|
||
atomic_int disarmed; /* set to 1 by ha_disarm_heartbeat_timeout_vtask */
|
||
pthread_t thread;
|
||
int started;
|
||
} hb_vtask_t;
|
||
|
||
/* File-scope pointer to the active heartbeat watchdog.
|
||
* Safety: arm/disarm/heartbeat are called only from the single-threaded
|
||
* vtaskmgr dispatch context; no locking is needed in the current architecture.
|
||
* If concurrent access is added later, protect with ha->lock. */
|
||
static hb_vtask_t *s_hb_vtask = NULL;
|
||
|
||
static void *hb_timeout_thread(void *arg)
|
||
{
|
||
hb_vtask_t *t = (hb_vtask_t *)arg;
|
||
ha_state_t *ha = t->ha;
|
||
|
||
while (!atomic_load(&t->disarmed)) {
|
||
/* Compute deadline: heartbeat_timeout_ms + rand_jitter.
|
||
* Must use heartbeat_timeout_ms (not election_timeout_ms) so standbys
|
||
* don't falsely trigger failover while the primary is in COMMITTING. */
|
||
uint32_t jitter_ms = 0;
|
||
if (ha->config.election_jitter_max_ms > 0) {
|
||
struct timespec now;
|
||
clock_gettime(CLOCK_REALTIME, &now);
|
||
unsigned int seed = (unsigned int)now.tv_nsec;
|
||
jitter_ms = (uint32_t)(rand_r(&seed) % ha->config.election_jitter_max_ms);
|
||
}
|
||
uint32_t total_ms = ha->config.heartbeat_timeout_ms + jitter_ms;
|
||
|
||
struct timespec deadline;
|
||
clock_gettime(CLOCK_REALTIME, &deadline);
|
||
deadline.tv_sec += (long)(total_ms / 1000);
|
||
deadline.tv_nsec += (long)(total_ms % 1000) * 1000000L;
|
||
if (deadline.tv_nsec >= 1000000000L) {
|
||
deadline.tv_sec++;
|
||
deadline.tv_nsec -= 1000000000L;
|
||
}
|
||
|
||
/* Wait for heartbeat signal or deadline */
|
||
int rc = sem_timedwait(&t->reset_sem, &deadline);
|
||
if (rc == -1 && errno == ETIMEDOUT) {
|
||
/* Timeout with no heartbeat → failover (Req 6.3, 6.4) */
|
||
if (!atomic_load(&t->disarmed)) {
|
||
WARN("HA: heartbeat timeout (%ums) → EPOLLRDHUP failover path",
|
||
total_ms);
|
||
ha_on_primary_ctrl_fd_rdhup(ha, t->c);
|
||
break; /* fired; thread exits */
|
||
}
|
||
}
|
||
/* heartbeat received — loop and re-arm deadline */
|
||
}
|
||
return NULL;
|
||
}
|
||
|
||
void ha_arm_heartbeat_timeout_vtask(ha_state_t *ha, cluster_t *c)
|
||
{
|
||
if (!ha) return;
|
||
|
||
/* Disarm any existing watchdog */
|
||
if (s_hb_vtask && s_hb_vtask->started) {
|
||
atomic_store(&s_hb_vtask->disarmed, 1);
|
||
sem_post(&s_hb_vtask->reset_sem); /* wake sleeping thread */
|
||
pthread_join(s_hb_vtask->thread, NULL);
|
||
sem_destroy(&s_hb_vtask->reset_sem);
|
||
free(s_hb_vtask);
|
||
s_hb_vtask = NULL;
|
||
}
|
||
|
||
hb_vtask_t *t = calloc(1, sizeof(*t));
|
||
if (!t) return;
|
||
|
||
t->ha = ha;
|
||
t->c = c;
|
||
t->started = 0;
|
||
atomic_init(&t->disarmed, 0);
|
||
sem_init(&t->reset_sem, 0, 0);
|
||
|
||
if (pthread_create(&t->thread, NULL, hb_timeout_thread, t) != 0) {
|
||
sem_destroy(&t->reset_sem);
|
||
free(t);
|
||
return;
|
||
}
|
||
t->started = 1;
|
||
s_hb_vtask = t;
|
||
}
|
||
|
||
void ha_on_heartbeat_received(ha_state_t *ha)
|
||
{
|
||
(void)ha;
|
||
if (s_hb_vtask && s_hb_vtask->started)
|
||
sem_post(&s_hb_vtask->reset_sem);
|
||
}
|
||
|
||
void ha_disarm_heartbeat_timeout_vtask(ha_state_t *ha)
|
||
{
|
||
(void)ha;
|
||
if (!s_hb_vtask) return;
|
||
if (s_hb_vtask->started) {
|
||
atomic_store(&s_hb_vtask->disarmed, 1);
|
||
sem_post(&s_hb_vtask->reset_sem);
|
||
pthread_join(s_hb_vtask->thread, NULL);
|
||
sem_destroy(&s_hb_vtask->reset_sem);
|
||
}
|
||
free(s_hb_vtask);
|
||
s_hb_vtask = NULL;
|
||
}
|
||
|
||
/* ── 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[] */
|
||
} standby_ep_ctx_t;
|
||
|
||
/*
|
||
* 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);
|
||
}
|
||
|
||
/* 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 */
|
||
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) {
|
||
int rc = xce_nop_req(ep);
|
||
if (rc < 0)
|
||
WARN("HA: heartbeat nop to standby[%d] failed: %d", i, rc);
|
||
}
|
||
}
|
||
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);
|
||
}
|
||
|