wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

feat(ha-single-leader-role): delete force-promotion watchdog + convergence/ack machinery (task 4.1)

Browse source
This commit is contained in:
Rongsong Shen 2026-06-13 12:05:08 +00:00
parent b7b951b9e7
commit a5bf86112a
10 changed files with 95 additions and 1869 deletions
Download patch file Download diff file Expand all files Collapse all files

218
src/controller/ha.c
View file

@ -61,13 +61,6 @@ typedef struct {
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_transport_only /
* ha_send_ctrlhello_to_all_standbys: defined in cluster.c / config_repl.c;
* only referenced in production builds.
@ -89,7 +82,6 @@ extern int push_config_transport_only(cluster_t *c, const uint8_t *bytes,
static void on_vote_connected(endpoint_t *ep);
static int vote_ep_err_handler(void *ep_raw);
static int committing_timeout_fn(void *arg);
/* ha-single-leader-role task 2.2: shared promotion seam — every road to
* PRIMARY registers the persist hooks and arms the own-term gate. */
static void ha_register_persist_hooks(cluster_t *c);
@ -637,18 +629,6 @@ static void raft_cb_become_follower(void *ctx, uint64_t term, int leader_idx)
/* Cancel any pending election timeout */
raft_cb_cancel_timeout(ctx);
/* Cancel any pending committing timeout — we are no longer leader.
* Without this, a node that stepped down due to a higher-term vote
* request would still promote to PRIMARY when the old committing
* timeout fires → split-brain. */
{
committing_timeout_ctx_t *ct = (committing_timeout_ctx_t *)
atomic_exchange(&ha->committing_timeout_ctx, NULL);
if (ct) {
atomic_store(&ct->cancelled, 1);
}
}
/* Sync ha->term */
atomic_store_explicit(&ha->term, term, memory_order_release);
@ -836,7 +816,6 @@ int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority)
ha->client_ssl_ctx = NULL;
ha->phase2_done = 0;
ha->n_peers = 0;
atomic_store_explicit(&ha->repl_ack_count, 0, memory_order_relaxed);
/* ha-heartbeat-liveness Task 1.1: zero-init liveness timestamp.
* Written by ha_on_heartbeat_received (task 2.1); read by liveness_poll_fn. */
@ -1295,67 +1274,6 @@ void ha_process_standby_state(ha_state_t *ha, int peer_idx,
}
}
/*
* ha_check_majority_convergence — evaluate whether a majority of standbys have
* converged and promote to CTRL_PRIMARY if so (Task 5.2).
*
* Counts the number of converged standbys from ha->standby_converged[].
* The majority threshold is ha->n_peers (all standbys must converge).
* For single-node clusters (n_peers=0), the threshold is 0, satisfied immediately.
*
* Only acts when role == CTRL_PRIMARY_COMMITTING. Called from the single-threaded
* vtaskmgr/iomgr context so no additional lock is needed; the role check is
* sufficient to prevent double-promotion.
*
* Logs INFO "majority converged (N/M standbys) → CTRL_PRIMARY" (Req 7.4, 2.4).
* Calls ha_promote_to_primary(c) which increments term and registers hooks.
* When c == NULL (unit tests with zeroed cluster_t), calls ha_do_promote(ha)
* directly so role transitions correctly without a live cluster.
*
* Requirements: 2.3, 2.4, 6.3, 6.5 (ha-repl-hardening spec)
*/
void ha_check_majority_convergence(ha_state_t *ha, cluster_t *c)
{
if (!ha) return;
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
if (role != CTRL_PRIMARY_COMMITTING)
return;
/* Count converged standbys (skip self: indices 0 .. n_peers-2) */
int n_peers = ha->n_peers;
int n_converged = 0;
for (int i = 0; i < n_peers - 1 && i < MAX_HA_CANDIDATES; i++) {
if (ha->standby_converged[i])
n_converged++;
}
/* Majority threshold (Req 2.3): primary + converged standbys must form
* a majority of the cluster — ceil(n_peers/2) - 1 standbys, integer form
* n_peers / 2 (n=3 → 1, n=5 → 2). Safe: committed entries are already
* on a majority (per-entry ack) and safe_committed_lsn derives from a
* majority of voters; a lagging minority standby keeps catching up via
* the heartbeat-driven decision loop after promotion.
* Single-node (n_peers<=1): threshold=0 — satisfied immediately. */
int threshold = n_peers / 2;
if (n_converged < threshold)
return;
INFO("HA: majority converged (%d/%d standbys, threshold %d) → CTRL_PRIMARY",
n_converged, n_peers - 1, threshold);
if (c) {
/* Full promotion: increments term, sets role=PRIMARY, registers hooks,
* arms periodic heartbeat to standbys. */
ha_promote_to_primary(c);
} else {
/* Test mode (c == NULL): directly set role so assertions work without
* a live cluster_t. ha_do_promote increments term and sets PRIMARY. */
ha_do_promote(ha);
}
}
/*
* ha_dispatch_repl_action_for_standby — execute the decision-loop action for
* one standby on the current heartbeat cycle (Task 3.3).
@ -2156,12 +2074,12 @@ int handle_request_vote(ha_state_t *ha,
#endif /* XCE_TESTING */
/* ── Test-only hook shims (XCE_TESTING only) ──────────────────────────────
* Type definitions (election_timeout_ctx_t, committing_timeout_ctx_t),
* extern declarations (cluster_init_phase2, push_config_transport_only,
* Type definitions (election_timeout_ctx_t), extern declarations
* (cluster_init_phase2, push_config_transport_only,
* 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.
* (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
@ -2726,7 +2644,7 @@ done:
* repl_ack_decode_and_dispatch: decode an inbound packet on a promoted vote
* endpoint while role == CTRL_PRIMARY_COMMITTING or CTRL_PRIMARY.
*
* - OPCode_replCatchUpDone → ha_repl_ack_received
* - OPCode_replCatchUpDone → INFO no-op (ack-counting removed in task 4.1)
* - OPCode_reply with stateReportMsg → ha_process_standby_state (Task 3.1)
* - OPCode_reply with walackreplymsg → repl_on_ack (Task 4.1)
* - OPCode_reply with replconfigpushreplymsg → repl_on_config_ack
@ -2763,7 +2681,14 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
if (opcode == OPCode_replCatchUpDone) {
capn_free(&capn_ctx);
ha_repl_ack_received(ha, c);
/* ha-single-leader-role task 4.1: the force-promotion ack-counting
* machinery has been deleted — promotion is driven solely by the
* heartbeat-driven decision loop and per-entry majority ack. Standbys
* still send OPCode_replCatchUpDone and the dispatch gate still admits
* it; it is now an INFO-logged no-op. */
(void)ha;
(void)c;
INFO("HA: replCatchUpDone received (no-op; ack-counting machinery removed in task 4.1)");
} else if (opcode == OPCode_reply &&
outer_rq.body_which == Request_body_replymsg &&
outer_rq.body.replymsg.p.data) {
@ -2813,12 +2738,6 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
sr.config_tag, sr.applied_lsn,
committed_tag, committed_lsn);
/* Task 5.2: After updating convergence tracking, check whether a
* majority of standbys have converged and promote to CTRL_PRIMARY.
* Must run BEFORE the replication dispatch so that a fully-converged
* cluster enters PRIMARY before making further replication decisions. */
ha_check_majority_convergence(ha, c);
/* Derive oldest_wal_lsn from the WAL directory.
* The oldest retained segment's start LSN is the boundary —
* standbys below it need a snapshot (Task 3.5b). */
@ -3077,45 +2996,6 @@ int election_timeout_fn_for_test(ha_state_t *ha, ha_elect_ctx_t *ctx,
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) {
/* Count converged vs pending standbys for the WARN log (Req 7.5). */
int n_converged = 0;
int n_pending = 0;
int n_peers = ha->n_peers;
for (int i = 0; i < n_peers - 1; i++) {
if (ha->standby_converged[i])
n_converged++;
else
n_pending++;
}
WARN("HA: committing timeout — majority convergence not reached within timeout; "
"%d/%d standbys converged, %d still pending; "
"promoting to PRIMARY in degraded mode (Req 6.4, 7.5)",
n_converged, n_peers - 1, n_pending);
if (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}.
@ -3172,9 +3052,10 @@ int peer_ctrl_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt)
/*
* vote_ep_ctx_set_cluster_for_test: set the cluster_t pointer inside the
* vote_ep_ctx_t stored at ep->data. Tests that need ha_repl_ack_received
* to call ha_promote_to_primary(c) when quorum is reached must call this
* after vote_ep_ctx_build_for_test (which sets vctx->c = NULL by default).
* vote_ep_ctx_t stored at ep->data. Tests that need the promoted-endpoint
* dispatch path to reach ha_promote_to_primary(c) with a live cluster pointer
* must call this after vote_ep_ctx_build_for_test (which sets vctx->c = NULL
* by default).
* Requirements: task 7.3 (ha-peer-connection-reuse).
*/
void vote_ep_ctx_set_cluster_for_test(endpoint_t *ep, cluster_t *c)
@ -3208,69 +3089,6 @@ void raft_cb_become_follower_for_test(ha_state_t *ha, uint64_t term)
}
#endif /* XCE_TESTING */
/* ── Task 4.1 / Task 5.3: committing_timeout_fn — delaymgr callback ─────────
*
* Fires ha_promote_to_primary (degraded mode) if majority convergence has not
* been reached within committing_timeout_ms. The majority-convergence path
* (ha_check_majority_convergence) is the PRIMARY promotion mechanism; this
* timeout is the safety-net fallback (Req 6.4, 8.4).
*
* WARN log includes the number of standbys that have converged and the number
* still pending, satisfying Req 7.5.
*
* 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) {
/* Count converged vs pending standbys for the WARN log (Req 7.5). */
int n_converged = 0;
int n_pending = 0;
int n_peers = t->ha->n_peers;
for (int i = 0; i < n_peers - 1; i++) {
if (t->ha->standby_converged[i])
n_converged++;
else
n_pending++;
}
WARN("HA: committing timeout — majority convergence not reached within timeout; "
"%d/%d standbys converged, %d still pending; "
"promoting to PRIMARY in degraded mode (Req 6.4, 7.5)",
n_converged, n_peers - 1, n_pending);
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