lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(ha-config-bootstrap): complete quorum election with mutex safety (task 4.3)

Browse source 
- am_i_primary_quorum(): per-peer pthreads, connect_ssl_raw + OP_requestVote
- pthread_mutex_t lock on ha_elect_ctx_t guards votes_granted/n_replied/max_resp
- done_posted guard in ha_elect_evaluate_and_signal prevents double sem_post
- Replace am_i_primary() with am_i_primary_quorum() in myapp.c
- Delete ha_check_priority_stepdown and ha_check_heartbeat_nocluster
- INFO logs: election outcome (Req 10.1), FSM transitions (Req 10.5)
- 24 election tests + 24 startup tests, 0 failures
This commit is contained in:
Rongsong Shen 2026-05-28 08:28:35 +00:00
parent 0f88d5d32a
commit 0c9c720b03
4 changed files with 390 additions and 98 deletions
Download patch file Download diff file Expand all files Collapse all files

400
src/controller/ha.c
View file

@ -4,6 +4,7 @@
#include <arpa/inet.h>
#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <netdb.h>
#include <poll.h>
#include <pthread.h>
@ -130,6 +131,9 @@ void ha_do_promote(ha_state_t *ha)
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)
@ -138,6 +142,9 @@ void ha_do_demote(ha_state_t *ha)
/* 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 ─────────────────────────────────── */
@ -326,38 +333,7 @@ int ha_is_primary(const ha_state_t *ha)
== CTRL_PRIMARY;
}
/*
* ha_check_priority_stepdown: called from heartbeat handler on simultaneous boot
* (both nodes at term=0). Steps down PRIMARY to STANDBY if a higher-priority peer
* (lower index = lower sender_priority value) is observed.
*/
void ha_check_priority_stepdown(ha_state_t *ha, int sender_priority,
uint64_t sender_term)
{
/* Only act if sender has strictly higher priority (lower index) */
if (sender_priority >= ha->my_priority)
return;
uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire);
if (my_term > 0 || sender_term > 0)
return; /* term-based election took over — don't interfere */
/* Hold lock to make the role-check + demote atomic, preventing a
* concurrent promotion from racing between the two operations. */
pthread_mutex_lock(&ha->lock);
/* Re-check term under lock in case a concurrent promotion raced us */
my_term = atomic_load_explicit(&ha->term, memory_order_acquire);
if (my_term == 0 && sender_term == 0) {
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
if (role == CTRL_PRIMARY) {
WARN("HA: stepping down — higher-priority peer (priority=%d) at term=0",
sender_priority);
ha_do_demote(ha);
}
}
pthread_mutex_unlock(&ha->lock);
}
/* ha_check_priority_stepdown() deleted — replaced by quorum election (Req 2.1) */
/*
* start_heartbeat_listener / stop_heartbeat_listener:
@ -387,21 +363,17 @@ uint64_t ha_trigger_manual_failover_for_test(ha_state_t *ha)
}
#endif
/* ── Watchdog thread implementation ────────────────────────── */
/* 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);
/* Only standbys watch for primary heartbeat timeout */
if (role != CTRL_STANDBY) return;
/* If we have no primary endpoint, nothing to monitor */
if (ha->primary_host[0] == '\0') return;
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
pthread_mutex_lock(&ha->lock);
struct timespec last = ha->last_heartbeat;
pthread_mutex_unlock(&ha->lock);
@ -410,11 +382,10 @@ void ha_check_heartbeat_nocluster(ha_state_t *ha)
(uint64_t)(now.tv_sec - last.tv_sec) * 1000ULL +
(uint64_t)(now.tv_nsec - last.tv_nsec) / 1000000ULL;
if (elapsed_ms >= ha->config.heartbeat_timeout_ms) {
/* Primary appears dead — set failover_requested; watchdog acts on it */
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)
{
@ -431,7 +402,24 @@ static void *ha_watchdog_thread(void *arg)
/* Re-check stop flag after sleep */
if (atomic_load_explicit(&ha->watchdog_stop, memory_order_acquire)) break;
ha_check_heartbeat_nocluster(ha);
/* 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);
@ -823,6 +811,12 @@ ctrl_role_t ha_elect_evaluate_and_signal(ha_state_t *ha, ha_elect_ctx_t *ctx)
{
(void)ha; /* ha->term used for fencing in network path; not needed here */
/* Guard: called from handle_request_vote_reply under ctx->lock; called from
* unit-test path (peers==NULL) without threads — done_posted prevents double
* sem_post in the network path if two threads both hit quorum simultaneously. */
if (ctx->done_posted)
return (ctx->result == 1) ? CTRL_PRIMARY : CTRL_STANDBY;
int total_nodes = ctx->n_peers + 1; /* include self */
int quorum = (total_nodes + 1) / 2; /* ceil(total/2) */
@ -830,32 +824,37 @@ ctrl_role_t ha_elect_evaluate_and_signal(ha_state_t *ha, ha_elect_ctx_t *ctx)
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->result = 0;
ctx->done_posted = 1;
sem_post(&ctx->done);
return CTRL_STANDBY;
}
/* Check quorum */
if (ctx->votes_granted < quorum) {
ctx->result = 0;
ctx->result = 0;
ctx->done_posted = 1;
sem_post(&ctx->done);
return CTRL_STANDBY;
}
/* Req 2.7: yield gate — fresher responding peer */
if (ctx->max_resp_tag > my_tag) {
ctx->result = 0;
ctx->result = 0;
ctx->done_posted = 1;
sem_post(&ctx->done);
return CTRL_STANDBY;
}
if (ctx->max_resp_tag == my_tag && ctx->max_resp_lsn > my_lsn) {
ctx->result = 0;
ctx->result = 0;
ctx->done_posted = 1;
sem_post(&ctx->done);
return CTRL_STANDBY;
}
/* Quorum met and no fresher peer → PRIMARY */
ctx->result = 1;
ctx->result = 1;
ctx->done_posted = 1;
sem_post(&ctx->done);
return CTRL_PRIMARY;
}
@ -890,11 +889,15 @@ void handle_request_vote_reply(ha_state_t *ha,
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)
if (entry->replied) {
pthread_mutex_unlock(&ctx->lock);
return;
}
/* Record reply fields */
entry->replied = 1;
@ -917,12 +920,215 @@ void handle_request_vote_reply(ha_state_t *ha,
/* 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. */
* n_peers is the peer count excluding self; self-vote is already in votes_granted.
* ha_elect_evaluate_and_signal is called under ctx->lock; its done_posted guard
* prevents double sem_post if two threads reach quorum simultaneously. */
int total_nodes = ctx->n_peers + 1;
int quorum = (total_nodes + 1) / 2;
if (ctx->votes_granted >= quorum || ctx->n_replied >= ctx->n_peers)
ha_elect_evaluate_and_signal(ha, ctx);
pthread_mutex_unlock(&ctx->lock);
}
/* ── Per-peer vote-request thread ──────────────────────────────────────────
*
* vote_thread_ctx_t: arguments passed to each vote_request_thread().
* One struct per peer; allocated on the stack of am_i_primary_quorum.
*/
typedef struct {
ha_state_t *ha; /* Candidate's HA state (term, ssl_ctx) */
ha_elect_ctx_t *ctx; /* Shared election context */
xce_candidate_t *peer; /* Target peer */
int peer_idx; /* Index in ctx->votes[] */
uint64_t my_term; /* Term sent in the request */
uint64_t my_tag; /* config_tag sent in the request */
uint64_t my_lsn; /* WAL LSN sent in the request */
int timeout_ms; /* Connect + I/O timeout */
} vote_thread_ctx_t;
/*
* vote_request_thread: per-peer pthread that sends OP_requestVote and waits
* for a RequestVoteReply.
*
* Wire format (raw SSL, no iomgr):
* Send: uint64_t size_hdr + capnp(RequestVoteRequest)
* Receive: uint64_t size_hdr + capnp(RequestVoteReply)
*
* On any error (connect, encode, I/O, decode) the peer counts as denied with
* zero peer_tag/peer_lsn — the quorum evaluator sees an absent peer as a no-vote.
*/
static void *vote_request_thread(void *arg)
{
vote_thread_ctx_t *vctx = (vote_thread_ctx_t *)arg;
ha_state_t *ha = vctx->ha;
ha_elect_ctx_t *ctx = vctx->ctx;
int idx = vctx->peer_idx;
SSL *ssl = NULL;
int sock = -1;
/* ── 1. TCP connect (non-blocking poll loop) ─────────── */
struct addrinfo hints, *res = NULL;
memset(&hints, 0, sizeof(hints));
hints.ai_family = AF_UNSPEC;
hints.ai_socktype = SOCK_STREAM;
char port_str[16];
snprintf(port_str, sizeof(port_str), "%d", vctx->peer->port);
if (getaddrinfo(vctx->peer->hostname, port_str, &hints, &res) != 0)
goto deny;
sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (sock < 0) { freeaddrinfo(res); sock = -1; goto deny; }
{
int flags = fcntl(sock, F_GETFL, 0);
fcntl(sock, F_SETFL, flags | O_NONBLOCK);
connect(sock, res->ai_addr, res->ai_addrlen);
freeaddrinfo(res); res = NULL;
struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP };
int pr;
do { pr = poll(&pfd, 1, vctx->timeout_ms); } while (pr < 0 && errno == EINTR);
if (pr <= 0 || (pfd.revents & (POLLERR | POLLHUP))) goto deny;
int err = 0; socklen_t el = sizeof(err);
getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
if (err != 0) goto deny;
/* Restore blocking for SSL handshake */
fcntl(sock, F_SETFL, flags & ~O_NONBLOCK);
}
/* ── 2. SSL handshake ────────────────────────────────── */
if (ha->ssl_ctx) {
ssl = SSL_new((SSL_CTX *)ha->ssl_ctx);
if (!ssl) goto deny;
SSL_set_fd(ssl, sock);
int sr;
do {
sr = SSL_connect(ssl);
if (sr <= 0) {
int e = SSL_get_error(ssl, sr);
if (e != SSL_ERROR_WANT_READ && e != SSL_ERROR_WANT_WRITE) goto deny;
}
} while (sr <= 0);
if (sr <= 0) goto deny;
}
/* ── 3. Encode + send RequestVoteRequest ─────────────── */
{
struct capn c;
struct capn_segment *cs;
struct RequestVoteRequest req_s;
RequestVoteRequest_ptr req_p;
uint8_t *buf = NULL;
int64_t blen = 0;
capn_init_malloc(&c);
cs = capn_root(&c).seg;
memset(&req_s, 0, sizeof(req_s));
req_s.term = vctx->my_term;
req_s.candidateId.str = ""; req_s.candidateId.len = 0; /* no name needed here */
req_s.configTag = vctx->my_tag;
req_s.lastLogLsn = vctx->my_lsn;
req_p = new_RequestVoteRequest(cs);
write_RequestVoteRequest(&req_s, req_p);
capn_setp(capn_root(&c), 0, req_p.p);
blen = capn_size(&c);
if (blen > 0) buf = malloc((size_t)blen);
if (!buf || capn_write_mem(&c, buf, (size_t)blen, 0) < 0) {
free(buf); capn_free(&c); goto deny;
}
capn_free(&c);
uint64_t hdr = (uint64_t)blen;
int nsent;
if (ssl) {
nsent = SSL_write(ssl, &hdr, (int)sizeof(hdr));
if (nsent == (int)sizeof(hdr))
nsent = SSL_write(ssl, buf, (int)blen);
} else {
nsent = (int)write(sock, &hdr, sizeof(hdr));
if (nsent == (int)sizeof(hdr))
nsent = (int)write(sock, buf, (size_t)blen);
}
free(buf);
if (nsent != (int)blen) goto deny;
}
/* ── 4. Read + decode RequestVoteReply ───────────────── */
{
uint64_t hdr = 0;
size_t off = 0;
/* Read the 8-byte size prefix */
while (off < sizeof(hdr)) {
int n;
if (ssl)
n = SSL_read(ssl, (char *)&hdr + off, (int)(sizeof(hdr) - off));
else
n = (int)read(sock, (char *)&hdr + off, sizeof(hdr) - off);
if (n <= 0) goto deny;
off += (size_t)n;
}
if (hdr == 0 || hdr > 65536) goto deny; /* sanity cap */
uint8_t *buf = malloc((size_t)hdr);
if (!buf) goto deny;
off = 0;
while (off < (size_t)hdr) {
int n;
if (ssl)
n = SSL_read(ssl, (char *)buf + off, (int)((size_t)hdr - off));
else
n = (int)read(sock, (char *)buf + off, (size_t)hdr - off);
if (n <= 0) { free(buf); goto deny; }
off += (size_t)n;
}
struct capn c;
struct RequestVoteReply rep_s;
RequestVoteReply_ptr rep_p;
if (capn_init_mem(&c, buf, (size_t)hdr, 0) != 0) {
free(buf); goto deny;
}
rep_p.p = capn_getp(capn_root(&c), 0, 1);
read_RequestVoteReply(&rep_s, rep_p);
capn_free(&c);
free(buf);
/* Fencing: if peer's term > ours, step down immediately */
if (rep_s.term > vctx->my_term)
atomic_store_explicit(&ha->term, rep_s.term, memory_order_release);
/* Record the reply */
handle_request_vote_reply(ha, ctx, idx,
rep_s.term,
rep_s.voteGranted ? 1 : 0,
rep_s.peerConfigTag,
rep_s.peerLastLogLsn);
}
goto cleanup;
deny:
/* Treat unreachable / error peer as denied vote with zero freshness */
handle_request_vote_reply(ha, ctx, idx,
vctx->my_term, /* use our term so no fencing */
0, /* vote_granted = false */
0, 0);
cleanup:
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
if (sock >= 0) close(sock);
return NULL;
}
/* ── Top-Level Election Entrypoint (Req 2.1, 2.9, 2.10, 2.11) ──────────────
@ -935,8 +1141,8 @@ void handle_request_vote_reply(ha_state_t *ha,
* Split vote → one retry with incremented term and randomized jitter (Req 2.10).
* Second failure → CTRL_STANDBY (Req 2.11).
*
* NOTE: In unit tests peers is NULL and n_peers==1; no network I/O occurs.
* Real multi-peer network path is tested via integration tests.
* Unit tests: peers==NULL and n_peers==1 exercises the single-node shortcircuit.
* Multi-peer network path (peers!=NULL) is exercised by integration tests.
*/
ctrl_role_t am_i_primary_quorum(ha_state_t *ha,
int n_peers,
@ -947,15 +1153,27 @@ ctrl_role_t am_i_primary_quorum(ha_state_t *ha,
/* Req 2.9: single-node self-elects immediately */
if (n_peers == 1) {
atomic_fetch_add_explicit(&ha->term, 1, memory_order_acq_rel);
uint64_t term = atomic_load_explicit(&ha->term, memory_order_acquire);
INFO("HA election: outcome=PRIMARY term=%" PRIu64
" votes_granted=1 quorum=1 (single-node)", term);
INFO("HA FSM: CTRL_INIT → CTRL_PRIMARY (single-node, term=%" PRIu64 ")", term);
return CTRL_PRIMARY;
}
/* Multi-node path: up to 2 election rounds (Req 2.10, 2.11) */
int n_other = n_peers - 1; /* number of peers excluding self */
int total_nodes = n_peers;
int quorum = (total_nodes + 1) / 2;
for (int round = 0; round < 2; round++) {
/* Increment term before each election round */
atomic_fetch_add_explicit(&ha->term, 1, memory_order_acq_rel);
uint64_t cur_term = atomic_load_explicit(&ha->term, memory_order_acquire);
/* Req 10.5: log FSM transition to CTRL_ELECTING */
INFO("HA FSM: → CTRL_ELECTING (round=%d, term=%" PRIu64 ", n_peers=%d)",
round + 1, cur_term, n_peers);
atomic_store_explicit(&ha->role, (int32_t)CTRL_ELECTING, memory_order_release);
/* Add jitter delay between rounds (Req 2.10) */
if (round > 0 && ha->config.election_jitter_max_ms > 0) {
@ -984,6 +1202,8 @@ ctrl_role_t am_i_primary_quorum(ha_state_t *ha,
ctx->max_resp_tag = 0;
ctx->max_resp_lsn = 0;
ctx->result = -1;
ctx->done_posted = 0;
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
/* Store candidate's own tag/lsn in votes[0] for evaluate function */
@ -992,21 +1212,73 @@ ctrl_role_t am_i_primary_quorum(ha_state_t *ha,
ctx->votes[0].my_lsn = my_lsn;
}
/* In real multi-peer path: spawn per-peer threads here (integration test path).
* For unit tests peers==NULL and n_other>0 means we're testing network path
* which is not mocked. Skip thread spawning when peers is NULL. */
if (peers != NULL) {
/* Production path: spawn vote request threads (not implemented in unit test scope).
* Threads would call handle_request_vote on peers and post to ctx->done.
* Integration tests exercise this path. */
(void)peers;
/* Spawn per-peer vote-request threads when peers list is provided.
* When peers==NULL (unit test mode), skip spawning and evaluate
* immediately with only the self-vote. */
vote_thread_ctx_t *vctxs = NULL;
pthread_t *tids = NULL;
if (peers != NULL && n_other > 0) {
vctxs = calloc((size_t)n_other, sizeof(*vctxs));
tids = calloc((size_t)n_other, sizeof(*tids));
if (!vctxs || !tids) {
free(vctxs); free(tids);
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
free(ctx);
return CTRL_STANDBY;
}
for (int i = 0; i < n_other; i++) {
vctxs[i].ha = ha;
vctxs[i].ctx = ctx;
vctxs[i].peer = peers[i];
vctxs[i].peer_idx = i;
vctxs[i].my_term = cur_term;
vctxs[i].my_tag = my_config_tag;
vctxs[i].my_lsn = my_lsn;
vctxs[i].timeout_ms =
(int)ha->config.probe_connect_timeout_ms;
pthread_create(&tids[i], NULL, vote_request_thread, &vctxs[i]);
}
/* Block until quorum or all peers replied */
sem_wait(&ctx->done);
/* Join all threads before freeing ctx */
for (int i = 0; i < n_other; i++)
pthread_join(tids[i], NULL);
free(vctxs);
free(tids);
}
/* If no peers to query (or no threads spawned), evaluate with self-vote only.
* For unit tests with n_other>0 but peers==NULL: no replies arrive.
* Evaluate with whatever is in ctx (self-vote only → may not reach quorum). */
ctrl_role_t role = ha_elect_evaluate_and_signal(ha, ctx);
ctrl_role_t role;
if (peers == NULL) {
/* Unit-test path: no network peers; evaluate with self-vote only */
role = ha_elect_evaluate_and_signal(ha, ctx);
} else {
/* Production path: ctx->result was set by ha_elect_evaluate_and_signal
* inside handle_request_vote_reply; just read it back. */
role = (ctx->result == 1) ? CTRL_PRIMARY : CTRL_STANDBY;
}
/* Capture before freeing ctx */
int votes_granted_snap = ctx->votes_granted;
/* Req 10.1: log election outcome */
INFO("HA election: outcome=%s term=%" PRIu64
" votes_granted=%d quorum=%d (round=%d)",
(role == CTRL_PRIMARY) ? "PRIMARY" : "STANDBY",
cur_term, votes_granted_snap, quorum, round + 1);
/* Req 10.5: FSM log for multi-node CTRL_PRIMARY path.
* ha_do_promote() is not called here so its FSM log is absent; emit it now. */
if (role == CTRL_PRIMARY)
INFO("HA FSM: → CTRL_PRIMARY (term=%" PRIu64 " round=%d votes=%d quorum=%d)",
cur_term, round + 1, votes_granted_snap, quorum);
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
free(ctx);