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

fix(ha-bootstrap): three startup fixes for election reliability

Browse source 
1. iomgr_monitor double-register: cluster_start skips iomgr_monitor
   if c->ep was already registered by cluster_register_listener
   (ep->flags non-zero after successful registration). Previously
   returned EEXIST → -1 → "Operation not permitted" crash.

2. ASAN stack-use-after-return: cluster_fini now calls iomgr_stop()
   before iomgr_fini() to join the worker thread before the cluster_t
   stack frame is reclaimed by the caller.

3. Election timing: add ha_wait_for_heartbeat_or_timeout() before
   election in myapp_run. If an existing primary sends a heartbeat,
   node becomes STANDBY immediately without competing in election.
   If timeout expires (no primary running), proceed to election.
   This prevents ctrl-a from immediately losing when ctrl-b/c are
   not yet started — it now waits election_timeout_ms (2s) first.
This commit is contained in:
Rongsong Shen 2026-05-30 00:28:10 +00:00
parent fc77630e6c
commit 47c09d653c
2 changed files with 38 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

11
src/controller/cluster.c
View file

@ -1119,10 +1119,12 @@ int cluster_start(cluster_t* c)
endpoint_set_conn_handler(&(c->ep), (conn_handler_t)handle_connection);
c->ep.iomgr = &(c->iomgr);
rc = iomgr_monitor(&(c->ep));
if (rc < 0) {
return rc;
/* Skip iomgr_monitor if cluster_register_listener already registered
* this fd with epoll (ep->flags is set on successful registration). */
if (c->ep.flags == 0) {
rc = iomgr_monitor(&(c->ep));
if (rc < 0)
return rc;
}
/* Guard: myapp_run() starts iomgr and vtaskmgr before election so that
@ -1207,6 +1209,7 @@ void cluster_fini(cluster_t* c)
xce_run_hooks(&(c->hooksys), XCE_CONTROLLER_HOOK_CLUSTER, HOOK_AFTER_EVENT,
HOOK_DETAIL_CLUSTER_SHUTDOWN, c);
cbuf_fini(&(c->wbuf));
iomgr_stop(&(c->iomgr)); /* join worker thread before freeing SSL_CTX */
iomgr_fini(&(c->iomgr));
destroy(c->jobmgr);

62
src/controller/myapp.c
View file

@ -113,15 +113,27 @@ int myapp_run(myconf_t* myconf)
return rc;
}
/* Election: quorum-based role determination (Req 2.1, 2.9).
* Build the peer pointer array (excluding self) for am_i_primary_quorum().
* peers_cfg.candidates includes this node; n_peers = candidates_count.
* my_priority is the index of this node in the candidates list. */
/* Req 1.1: wait for an incoming heartbeat before starting election.
* If a primary already exists it will send heartbeats; receiving one means
* this node should become STANDBY immediately (skip election).
* If no heartbeat arrives within election_timeout_ms + jitter, proceed to
* election — no primary is reachable and this node should compete. */
int skip_election = 0;
{
ctrl_role_t hb_role = ha_wait_for_heartbeat_or_timeout(&c.ha);
if (hb_role == CTRL_STANDBY) {
/* A primary is already running — skip election, go straight to standby */
INFO("HA: heartbeat received — existing primary detected, becoming STANDBY");
skip_election = 1;
}
/* CTRL_ELECTING: no heartbeat received, proceed to election */
}
if (!skip_election) {
/* Election: quorum-based role determination (Req 2.1, 2.9). */
const xce_cluster_config_t *pcfg = &myconf->peers_cfg;
int n_peers = (pcfg->candidates_count > 0) ? pcfg->candidates_count : 1;
/* Build array of pointers to other candidates (excluding self by index) */
xce_candidate_t *peer_ptrs[MAX_HA_CANDIDATES];
int n_other = 0;
for (int i = 0; i < pcfg->candidates_count; i++) {
@ -134,46 +146,25 @@ int myapp_run(myconf_t* myconf)
memory_order_acquire);
ctrl_role_t elected = am_i_primary_quorum(
&c.ha,
n_peers,
(n_other > 0) ? peer_ptrs : NULL,
my_tag,
my_lsn);
&c.ha, n_peers, (n_other > 0) ? peer_ptrs : NULL, my_tag, my_lsn);
if (elected == CTRL_PRIMARY) {
/* Primary: load config from conf/current (FATAL if missing) */
rc = load_primary_config(&c);
if (rc != 0) {
cluster_fini(&c);
myconf_fini(myconf);
return rc;
}
/* am_i_primary_quorum logs FSM transition; set role explicitly */
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY,
memory_order_release);
} else {
/* Req 3.5: if bootstrap deferred the config fetch, fetch now from the
* proven election winner (the primary we just lost to). */
if (c.ha.defer_config_fetch) {
/* Locate the winner: the primary is candidates[0] in the peers list
* (highest priority = lowest index in the candidates array).
* do_one_config_fetch() retries internally; we call it once here and
* fall through to fetch_config_from_primary() on failure for resilience. */
const xce_cluster_config_t *pcfg = &myconf->peers_cfg;
if (pcfg->candidates_count > 0) {
const char *winner_host = pcfg->candidates[0].hostname;
uint32_t winner_port = pcfg->candidates[0].port;
const xce_cluster_config_t *p2 = &myconf->peers_cfg;
if (p2->candidates_count > 0)
INFO("HA bootstrap: deferred config fetch from winner %s:%u",
winner_host, (unsigned)winner_port);
/* do_one_config_fetch is static in config_repl.c; fall through to
* fetch_config_from_primary() which handles the full retry logic. */
}
p2->candidates[0].hostname, (unsigned)p2->candidates[0].port);
c.ha.defer_config_fetch = false;
}
/* Standby: fetch config from primary before Phase 2.
* Skip if this node already has a valid local config (ground-truth
* nodes should not fetch from primary — they ARE the ground truth). */
if (myconf->config_version_tag == 0) {
rc = fetch_config_from_primary(&c);
if (rc != 0) {
@ -185,7 +176,16 @@ int myapp_run(myconf_t* myconf)
INFO("Standby has local config (tag=%016llx) — skipping fetch from primary",
(unsigned long long)myconf->config_version_tag);
}
/* Role stays CTRL_STANDBY — already set by ha_init(). */
}
} else {
/* Heartbeat-detected STANDBY: fetch config if not already loaded. */
if (myconf->config_version_tag == 0) {
rc = fetch_config_from_primary(&c);
if (rc != 0) {
cluster_fini(&c);
myconf_fini(myconf);
return rc;
}
}
}