fix(ha-election,ha-rejoin): stop two infinite-election loops on failover and restart
Two distinct HA livelocks, both verified on a live 3-node ASAN deploy (kill+restart of primary and standby, plus SIGSTOP wedge injection). bug-339 — election freshness-floor asymmetry (RDHUP failover path): After the primary dies the two survivors could never elect a leader — both only ever became follower while terms climbed in lockstep. The RDHUP candidate advertised c->persist.committed_lsn, which a standby never advances (only the primary commit engine writes it), while every voter computed its freshness floor from max(applied_lsn, last_known_committed_lsn). So each survivor advertised a stale-low LSN against a high voter floor and the freshness gate denied on both sides. Fix: compute the candidate LSN via ha_vote_self_freshness so the advertised value equals the voter floor (ha_on_primary_ctrl_fd_rdhup); also route the boot-path election (myapp_run) through the same helper for symmetry. bug-340 — wedged standby endpoint suppressed reconnect (rejoin path): A killed/restarted peer's outbound endpoint on the primary could wedge (epoll re-arm ENOENT) yet stay in standby_endpoints, so primary_heartbeat_fn's reconnect scan treated the peer as still-connected and never re-established a ctrlhello channel — the restarted node never heard from the primary and re-elected every heartbeat_timeout (term climbing unbounded). xce_nop_req returns success on enqueue even when the send can't progress, so the heartbeat layer never noticed. Fix: primary_heartbeat_fn now evicts a standby that was responsive once (ts_last_in > 0) but has been silent past heartbeat_timeout — symmetric with the standby's own liveness poll — and the reconnect scan re-establishes the channel. Uses existing endpoint_t.ts_last_in (no libnetio change). Tests (RED->GREEN): - test_ha_rdhup_integration: candidate advertises freshness floor, not stale committed_lsn - test_ha_liveness: evicts_silent_standby + keeps_live/keeps_unconfirmed (no false eviction)
This commit is contained in:
parent
ca1d9ae42d
commit
a735a880ae
4 changed files with 271 additions and 10 deletions
|
|
@ -725,7 +725,8 @@ static void raft_cb_become_follower(void *ctx, uint64_t term, int leader_idx)
|
|||
}
|
||||
|
||||
/* Arm heartbeat watchdog — fires ha_on_primary_ctrl_fd_rdhup if primary
|
||||
* sends no ctrlhello within election_timeout_ms. */
|
||||
* sends no ctrlhello within election_timeout_ms. (This also resets
|
||||
* last_activity = now, granting a full grace window before the first poll.) */
|
||||
ha_arm_heartbeat_timeout_vtask(ha, c);
|
||||
}
|
||||
|
||||
|
|
@ -3557,6 +3558,28 @@ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c)
|
|||
* so that the subsequent raft_start_election finds a valid context. */
|
||||
ha->raft_c = c;
|
||||
|
||||
/* Candidate freshness for the RequestVote payload. This MUST mirror the
|
||||
* voter-side freshness floor (cluster.c OPCode_requestVote): the effective
|
||||
* floor is max(applied_lsn, last_known_committed_lsn) at the EFFECTIVE
|
||||
* config tag. A standby never advances persist.committed_lsn (only the
|
||||
* primary's commit engine writes it) — it advances applied_lsn (WAL
|
||||
* catch-up) and last_known_committed_lsn (heartbeat replies). Advertising
|
||||
* persist.committed_lsn here makes every surviving standby advertise a
|
||||
* stale-low LSN while each peer computes a HIGH floor from applied_lsn;
|
||||
* the freshness gate (cand_lsn < my_lsn) then trips on both sides, each
|
||||
* denies the other's vote, no candidate reaches quorum, and the cluster
|
||||
* spins in an infinite RDHUP→ELECTING→STANDBY loop after the primary dies.
|
||||
* Using ha_vote_self_freshness makes advertised == voter-floor, so the
|
||||
* freshest node is always granted by every peer. (Matches the boot-path
|
||||
* election in myapp_run, which advertises applied_lsn.) */
|
||||
uint64_t cand_tag = 0, cand_lsn = 0;
|
||||
if (c->myconf) {
|
||||
uint64_t local_tag = c->myconf->config_version_tag;
|
||||
uint64_t local_lsn = atomic_load_explicit(&ha->applied_lsn,
|
||||
memory_order_acquire);
|
||||
ha_vote_self_freshness(ha, local_tag, local_lsn, &cand_tag, &cand_lsn);
|
||||
}
|
||||
|
||||
/* Allocate fresh elect_ctx so raft_cb_send_vote_request can read my_tag/my_lsn */
|
||||
if (c && c->myconf) {
|
||||
int n_peers = c->myconf->peers_cfg.candidates_count;
|
||||
|
|
@ -3564,9 +3587,8 @@ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c)
|
|||
ha_elect_ctx_t *new_ctx = calloc(1, sizeof(*new_ctx) +
|
||||
(size_t)n_other * sizeof(new_ctx->votes[0]));
|
||||
if (new_ctx) {
|
||||
uint64_t my_tag = c->myconf->config_version_tag;
|
||||
uint64_t my_lsn = atomic_load_explicit(&c->persist.committed_lsn,
|
||||
memory_order_acquire);
|
||||
uint64_t my_tag = cand_tag;
|
||||
uint64_t my_lsn = cand_lsn;
|
||||
new_ctx->n_peers = n_other;
|
||||
new_ctx->votes_granted = 1; /* self-vote placeholder */
|
||||
new_ctx->result = -1;
|
||||
|
|
@ -3607,9 +3629,8 @@ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c)
|
|||
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);
|
||||
uint64_t my_tag = cand_tag;
|
||||
uint64_t my_lsn = cand_lsn;
|
||||
const char *my_name = c->myconf ? c->myconf->name : "";
|
||||
ha_rdhup_election_hook(ha, c, np, peer_ptrs, my_tag, my_lsn, my_name);
|
||||
}
|
||||
|
|
@ -4094,6 +4115,17 @@ static int primary_heartbeat_fn(void *arg)
|
|||
interval_s = HA_DEFAULT_HEARTBEAT_INTERVAL_MS / 1000;
|
||||
|
||||
if (ha->standby_endpoints) {
|
||||
/* Liveness eviction threshold (wall clock, matches ep->ts_last_in). */
|
||||
time_t now_wall = time(NULL);
|
||||
uint32_t timeout_s = ha->config.heartbeat_timeout_ms / 1000;
|
||||
if (timeout_s == 0)
|
||||
timeout_s = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS / 1000;
|
||||
|
||||
/* Endpoints evicted this tick; their repl deregistration runs AFTER
|
||||
* ha->lock is released (repl lock-ordering — mirrors the err handlers). */
|
||||
endpoint_t *evicted[MAX_HA_CANDIDATES];
|
||||
int n_evicted = 0;
|
||||
|
||||
pthread_mutex_lock(&ha->lock);
|
||||
int n = ha->standby_endpoints->count;
|
||||
for (int i = 0; i < n; i++) {
|
||||
|
|
@ -4101,6 +4133,30 @@ static int primary_heartbeat_fn(void *arg)
|
|||
if (!ep) continue;
|
||||
standby_ep_ctx_t *sctx = (standby_ep_ctx_t *)ep->data;
|
||||
if (!sctx) continue;
|
||||
|
||||
/* Liveness eviction (symmetric with the standby's liveness poll):
|
||||
* a standby that was responsive at least once (ts_last_in > 0) but
|
||||
* has sent nothing for longer than heartbeat_timeout is dead or was
|
||||
* restarted. A wedged/dead endpoint left in the array makes the
|
||||
* reconnect scan below treat the peer as still-connected, so the
|
||||
* restarted node never receives a fresh ctrlhello and loops
|
||||
* elections forever (terms climbing unbounded). Evict it here; the
|
||||
* reconnect scan re-establishes a fresh channel on this/next tick.
|
||||
* Removal from standby_endpoints is the load-bearing action — the
|
||||
* endpoint object itself is reclaimed by its err handler when the
|
||||
* dead connection finally surfaces an epoll event (bounded). */
|
||||
if (ep->ts_last_in > 0 &&
|
||||
(now_wall - ep->ts_last_in) > (time_t)timeout_s) {
|
||||
WARN("HA: standby[%d] silent for %lds (timeout=%us) — evicting "
|
||||
"dead endpoint, will reconnect",
|
||||
i, (long)(now_wall - ep->ts_last_in), timeout_s);
|
||||
zarray_del(ha->standby_endpoints, i);
|
||||
if (n_evicted < MAX_HA_CANDIDATES)
|
||||
evicted[n_evicted++] = ep;
|
||||
i--; n--;
|
||||
continue;
|
||||
}
|
||||
|
||||
time_t elapsed = now - sctx->last_push_time;
|
||||
if (elapsed >= (time_t)interval_s) {
|
||||
#ifdef XCE_TESTING
|
||||
|
|
@ -4115,6 +4171,9 @@ static int primary_heartbeat_fn(void *arg)
|
|||
}
|
||||
}
|
||||
pthread_mutex_unlock(&ha->lock);
|
||||
|
||||
for (int k = 0; k < n_evicted; k++)
|
||||
repl_remove_standby(&c->persist.repl, evicted[k]);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -213,9 +213,17 @@ int myapp_run(myconf_t *myconf)
|
|||
}
|
||||
}
|
||||
|
||||
uint64_t my_tag = myconf->config_version_tag;
|
||||
uint64_t my_lsn = (uint64_t)atomic_load_explicit(&c.ha.applied_lsn,
|
||||
memory_order_acquire);
|
||||
/* Advertise the EFFECTIVE freshness floor — identical to the voter-side
|
||||
* floor (cluster.c OPCode_requestVote) and to the RDHUP failover path
|
||||
* (ha_on_primary_ctrl_fd_rdhup): max(applied_lsn, last_known_committed_lsn)
|
||||
* at the effective config tag. Candidate-advertised LSN must equal what a
|
||||
* peer would compute as its voter floor, otherwise two equally-fresh nodes
|
||||
* mutually deny each other's votes and no leader is ever elected. */
|
||||
uint64_t local_tag = myconf->config_version_tag;
|
||||
uint64_t local_lsn = (uint64_t)atomic_load_explicit(&c.ha.applied_lsn,
|
||||
memory_order_acquire);
|
||||
uint64_t my_tag = 0, my_lsn = 0;
|
||||
ha_vote_self_freshness(&c.ha, local_tag, local_lsn, &my_tag, &my_lsn);
|
||||
|
||||
/* Initialize raft state machine once with cluster topology.
|
||||
* Must happen before any election or watchdog path that calls
|
||||
|
|
|
|||
|
|
@ -435,6 +435,117 @@ static void test_primary_hb_skips_nop_when_fresh(void)
|
|||
printf("test_primary_hb_skips_nop_when_fresh: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test 9b: primary evicts a standby that has gone silent (dead/restarted) ──
|
||||
*
|
||||
* Regression for the "kill+restart a controller → it loops elections forever"
|
||||
* bug. When a peer is killed/restarted, its outbound standby endpoint on the
|
||||
* primary can wedge (epoll re-arm fails) yet stay in standby_endpoints, so the
|
||||
* reconnect scan treats the peer as still-connected and never re-establishes a
|
||||
* fresh ctrlhello channel — the restarted node never hears from the primary and
|
||||
* re-elects every heartbeat_timeout (term climbing unbounded).
|
||||
*
|
||||
* Fix: primary_heartbeat_fn evicts any standby endpoint that was responsive once
|
||||
* (ts_last_in > 0) but has been silent longer than heartbeat_timeout, so the
|
||||
* reconnect scan re-establishes the channel. Before the fix the ep persists.
|
||||
*/
|
||||
static void test_primary_hb_evicts_silent_standby(void)
|
||||
{
|
||||
cluster_t c;
|
||||
memset(&c, 0, sizeof(c));
|
||||
init_ha_primary_with_standby_eps(&c.ha);
|
||||
|
||||
endpoint_t ep;
|
||||
memset(&ep, 0, sizeof(ep));
|
||||
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
|
||||
standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data;
|
||||
sctx->last_push_time = mono_now(); /* recently pushed — not the trigger */
|
||||
|
||||
/* Responsive once, then silent well beyond heartbeat_timeout (wall clock,
|
||||
* matching endpoint_t.ts_last_in which handle_input stamps via time(NULL)). */
|
||||
uint32_t timeout_s = g_cfg.heartbeat_timeout_ms / 1000;
|
||||
ep.ts_last_in = time(NULL) - (time_t)(timeout_s + 5);
|
||||
|
||||
zarray_store(c.ha.standby_endpoints, &ep);
|
||||
TEST_ASSERT_EQUAL_INT(1, c.ha.standby_endpoints->count);
|
||||
|
||||
g_nop_calls = 0;
|
||||
ha_nop_req_hook = stub_nop_req;
|
||||
primary_heartbeat_fn_for_test(&c);
|
||||
ha_nop_req_hook = NULL;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, c.ha.standby_endpoints->count,
|
||||
"primary must evict a standby silent beyond heartbeat_timeout so it reconnects");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_nop_calls,
|
||||
"no NOP should be sent to an evicted dead standby");
|
||||
|
||||
free(ep.data);
|
||||
fini_ha_primary_with_standby_eps(&c.ha);
|
||||
printf("test_primary_hb_evicts_silent_standby: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test 9c: a responsive standby is NOT evicted (no false positives) ─────── */
|
||||
static void test_primary_hb_keeps_live_standby(void)
|
||||
{
|
||||
cluster_t c;
|
||||
memset(&c, 0, sizeof(c));
|
||||
init_ha_primary_with_standby_eps(&c.ha);
|
||||
|
||||
endpoint_t ep;
|
||||
memset(&ep, 0, sizeof(ep));
|
||||
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
|
||||
standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data;
|
||||
sctx->last_push_time = mono_now();
|
||||
|
||||
/* Recent inbound activity — within the timeout window. */
|
||||
ep.ts_last_in = time(NULL);
|
||||
|
||||
zarray_store(c.ha.standby_endpoints, &ep);
|
||||
|
||||
g_nop_calls = 0;
|
||||
ha_nop_req_hook = stub_nop_req;
|
||||
primary_heartbeat_fn_for_test(&c);
|
||||
ha_nop_req_hook = NULL;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, c.ha.standby_endpoints->count,
|
||||
"a responsive standby (recent ts_last_in) must NOT be evicted");
|
||||
|
||||
free(ep.data);
|
||||
fini_ha_primary_with_standby_eps(&c.ha);
|
||||
printf("test_primary_hb_keeps_live_standby: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test 9d: a never-responded standby is NOT evicted (grace for new joiners) ── */
|
||||
static void test_primary_hb_keeps_unconfirmed_standby(void)
|
||||
{
|
||||
cluster_t c;
|
||||
memset(&c, 0, sizeof(c));
|
||||
init_ha_primary_with_standby_eps(&c.ha);
|
||||
|
||||
endpoint_t ep;
|
||||
memset(&ep, 0, sizeof(ep));
|
||||
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
|
||||
standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data;
|
||||
sctx->last_push_time = mono_now();
|
||||
|
||||
/* Freshly connected, no reply yet: ts_last_in == 0. Must NOT be evicted
|
||||
* (eviction keys on ts_last_in > 0 so a just-connected joiner gets grace). */
|
||||
ep.ts_last_in = 0;
|
||||
|
||||
zarray_store(c.ha.standby_endpoints, &ep);
|
||||
|
||||
g_nop_calls = 0;
|
||||
ha_nop_req_hook = stub_nop_req;
|
||||
primary_heartbeat_fn_for_test(&c);
|
||||
ha_nop_req_hook = NULL;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, c.ha.standby_endpoints->count,
|
||||
"a just-connected standby (ts_last_in==0) must NOT be evicted");
|
||||
|
||||
free(ep.data);
|
||||
fini_ha_primary_with_standby_eps(&c.ha);
|
||||
printf("test_primary_hb_keeps_unconfirmed_standby: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test 10: on_standby_connected sets last_push_time ──────────────────── */
|
||||
|
||||
/*
|
||||
|
|
@ -648,6 +759,9 @@ void test_ha_liveness(void)
|
|||
RUN_TEST(test_ha_disarm_makes_subsequent_poll_noop);
|
||||
RUN_TEST(test_primary_hb_sends_nop_when_idle);
|
||||
RUN_TEST(test_primary_hb_skips_nop_when_fresh);
|
||||
RUN_TEST(test_primary_hb_evicts_silent_standby);
|
||||
RUN_TEST(test_primary_hb_keeps_live_standby);
|
||||
RUN_TEST(test_primary_hb_keeps_unconfirmed_standby);
|
||||
RUN_TEST(test_on_standby_connected_sets_last_push_time);
|
||||
RUN_TEST(test_cluster_dispatch_nop_resets_activity_when_standby);
|
||||
RUN_TEST(test_cluster_dispatch_nop_skips_when_primary);
|
||||
|
|
|
|||
|
|
@ -496,10 +496,90 @@ static void test_rdhup_configured_node_still_elects(void)
|
|||
g_guard_hook_calls);
|
||||
}
|
||||
|
||||
/* ═══════════════════════════════════════════════════════════════════════════
|
||||
* Test 9 (regression — infinite election loop after primary death):
|
||||
* The RDHUP candidate must advertise its EFFECTIVE freshness floor
|
||||
* (max(applied_lsn, last_known_committed_lsn)) as the RequestVote last_log_lsn,
|
||||
* NOT the stale persist.committed_lsn.
|
||||
*
|
||||
* Root cause of the observed bug: a standby never advances
|
||||
* persist.committed_lsn (only the primary's commit engine writes it); it
|
||||
* advances applied_lsn (WAL catch-up) and last_known_committed_lsn (heartbeat
|
||||
* replies). The voter-side freshness floor (cluster.c OPCode_requestVote) is
|
||||
* max(applied_lsn, last_known_committed_lsn). When the RDHUP candidate
|
||||
* advertised persist.committed_lsn (stale-low) while every voter compared
|
||||
* against a high applied_lsn floor, the freshness gate (cand_lsn < my_lsn)
|
||||
* tripped on BOTH surviving standbys — each denied the other's vote, no
|
||||
* candidate ever reached quorum, and the cluster spun in an infinite
|
||||
* RDHUP→ELECTING→STANDBY loop with terms climbing in lockstep.
|
||||
*
|
||||
* Captured via the ha_rdhup_election_hook my_lsn argument.
|
||||
* Before the fix: captured == 0 (stale committed_lsn). After: == 10.
|
||||
* ═════════════════════════════════════════════════════════════════════════ */
|
||||
static uint64_t g_captured_my_lsn = 0;
|
||||
static uint64_t g_captured_my_tag = 0;
|
||||
|
||||
static void capture_lsn_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)
|
||||
{
|
||||
(void)ha; (void)c; (void)n_peers; (void)peers; (void)my_name;
|
||||
g_captured_my_tag = my_config_tag;
|
||||
g_captured_my_lsn = my_lsn;
|
||||
}
|
||||
|
||||
static void test_rdhup_candidate_advertises_freshness_floor_not_stale_committed(void)
|
||||
{
|
||||
install_stubs();
|
||||
g_captured_my_lsn = 0;
|
||||
g_captured_my_tag = 0;
|
||||
|
||||
cluster_t c;
|
||||
myconf_t myconf;
|
||||
make_cluster(&c, &myconf, "ctrl-a");
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &g_cfg, 0));
|
||||
ha_do_demote(&c.ha);
|
||||
c.ha.phase2_done = 1;
|
||||
c.ha.n_peers = 1;
|
||||
|
||||
/* Standby state after applying WAL + learning commits from the (now-dead)
|
||||
* primary: applied_lsn and last_known_committed_lsn advanced to 10, but
|
||||
* persist.committed_lsn is still 0 — a standby never advances it. */
|
||||
atomic_store_explicit(&c.ha.applied_lsn, 10ULL, memory_order_release);
|
||||
atomic_store_explicit(&c.ha.last_known_committed_lsn, 10ULL,
|
||||
memory_order_release);
|
||||
atomic_store_explicit(&c.persist.committed_lsn, 0ULL, memory_order_release);
|
||||
|
||||
ha_rdhup_election_hook = capture_lsn_hook;
|
||||
|
||||
int sv[2];
|
||||
TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv));
|
||||
c.ha.primary_ctrl_fd = sv[0];
|
||||
|
||||
ha_on_primary_ctrl_fd_rdhup(&c.ha, &c);
|
||||
|
||||
clear_stubs();
|
||||
close(sv[1]);
|
||||
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(10ULL, g_captured_my_lsn,
|
||||
"RDHUP candidate must advertise effective freshness floor "
|
||||
"(applied/learned-committed = 10), not stale persist.committed_lsn (0); "
|
||||
"otherwise surviving standbys mutually deny votes -> infinite election loop");
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE((uint64_t)TEST_CONFIG_TAG, g_captured_my_tag,
|
||||
"RDHUP candidate must advertise its config_version_tag");
|
||||
|
||||
ha_fini(&c.ha);
|
||||
printf("test_rdhup_candidate_advertises_freshness_floor_not_stale_committed: "
|
||||
"PASS (my_lsn=%llu)\n", (unsigned long long)g_captured_my_lsn);
|
||||
}
|
||||
|
||||
/* ── Suite entry point ──────────────────────────────────────────────────── */
|
||||
|
||||
void test_ha_rdhup_integration(void)
|
||||
{
|
||||
RUN_TEST(test_rdhup_candidate_advertises_freshness_floor_not_stale_committed);
|
||||
RUN_TEST(test_rdhup_full_path_reaches_primary);
|
||||
/* ack-driven promotion test deleted (ha-single-leader-role 4.1:
|
||||
* ack-driven promotion machinery removed, no shim) */
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue