diff --git a/src/controller/ha.c b/src/controller/ha.c index 1f407057..7c5b8978 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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]); } } diff --git a/src/controller/myapp.c b/src/controller/myapp.c index 5d7dda2a..940dbabe 100644 --- a/src/controller/myapp.c +++ b/src/controller/myapp.c @@ -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 diff --git a/src/controller/testsuite/test_ha_liveness.c b/src/controller/testsuite/test_ha_liveness.c index ff77a434..d16da9de 100644 --- a/src/controller/testsuite/test_ha_liveness.c +++ b/src/controller/testsuite/test_ha_liveness.c @@ -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); diff --git a/src/controller/testsuite/test_ha_rdhup_integration.c b/src/controller/testsuite/test_ha_rdhup_integration.c index b8b4e634..1d1fb19a 100644 --- a/src/controller/testsuite/test_ha_rdhup_integration.c +++ b/src/controller/testsuite/test_ha_rdhup_integration.c @@ -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) */