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

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

4
.kiro/specs/ha-single-leader-role/tasks.md
View file

@ -38,7 +38,7 @@
- _Requirements: 1.1, 1.5_ - _Requirements: 1.1, 1.5_
- [ ] 4. Machinery deletions (watchdog, convergence, ack counters, gate branch, predicates) - [ ] 4. Machinery deletions (watchdog, convergence, ack counters, gate branch, predicates)
- [ ] 4.1 Delete the force-promotion watchdog and ack/convergence machinery - [x] 4.1 Delete the force-promotion watchdog and ack/convergence machinery
- Delete from ha.c: `committing_timeout_fn` (3080-3121), `committing_timeout_fn_for_test` (2950-2990), `committing_timeout_ctx_t` type + fwd decl (64-69, 83), the cancel block in `raft_cb_become_follower` (640-648); `ha_check_majority_convergence` (1295-1353) and its heartbeat-reply call site (2710-2715); `ha_repl_ack_received` (3123-3141) — its `OPCode_replCatchUpDone` branch in `repl_ack_decode_and_dispatch` (2659-2661) becomes an INFO-logged no-op (opcode stays gate-allowed; `repl_complete_catchup` untouched); `repl_ack_count` init at ha.c:835 - Delete from ha.c: `committing_timeout_fn` (3080-3121), `committing_timeout_fn_for_test` (2950-2990), `committing_timeout_ctx_t` type + fwd decl (64-69, 83), the cancel block in `raft_cb_become_follower` (640-648); `ha_check_majority_convergence` (1295-1353) and its heartbeat-reply call site (2710-2715); `ha_repl_ack_received` (3123-3141) — its `OPCode_replCatchUpDone` branch in `repl_ack_decode_and_dispatch` (2659-2661) becomes an INFO-logged no-op (opcode stays gate-allowed; `repl_complete_catchup` untouched); `repl_ack_count` init at ha.c:835
- Delete declarations from ha.h: `ha_check_majority_convergence` (396-412), `ha_repl_ack_received` (626-630) - Delete declarations from ha.h: `ha_check_majority_convergence` (396-412), `ha_repl_ack_received` (626-630)
- Delete test files whose subject is this machinery: remove `testsuite/test_ha_repl_ack.c` and `testsuite/test_ha_evaluate_committing.c` and their entries in `controller_unit_test_src` (meson.build:116, 118 anchors); concerns still valid (promotion only on real quorum, no promotion without majority) are already pinned by task 2.1's gate tests - Delete test files whose subject is this machinery: remove `testsuite/test_ha_repl_ack.c` and `testsuite/test_ha_evaluate_committing.c` and their entries in `controller_unit_test_src` (meson.build:116, 118 anchors); concerns still valid (promotion only on real quorum, no promotion without majority) are already pinned by task 2.1's gate tests
@ -85,5 +85,7 @@
- **(2.2) ha.c shared sequence**: raft path stores `CTRL_PRIMARY` directly with NO `ha_do_promote` (term synced at entry — avoids double-increment); `ha_promote_to_primary` keeps `ha_do_promote` (owns its term). Both call static `ha_register_persist_hooks(c)` (BEFORE noop append) and static `ha_arm_own_term_gate(c)` (`persist_wal_async(WAL_NOOP)` → `noop_lsn = current_lsn-1` under p->lock → `commit_submit(RETAIN)` → `commit_set_writable_lsn`); append-failure degrades safe (gate unset, promotion completes); non-blocking. `committing_timeout_fn` + enum retained until tasks 4.x/5.2. - **(2.2) ha.c shared sequence**: raft path stores `CTRL_PRIMARY` directly with NO `ha_do_promote` (term synced at entry — avoids double-increment); `ha_promote_to_primary` keeps `ha_do_promote` (owns its term). Both call static `ha_register_persist_hooks(c)` (BEFORE noop append) and static `ha_arm_own_term_gate(c)` (`persist_wal_async(WAL_NOOP)` → `noop_lsn = current_lsn-1` under p->lock → `commit_submit(RETAIN)` → `commit_set_writable_lsn`); append-failure degrades safe (gate unset, promotion completes); non-blocking. `committing_timeout_fn` + enum retained until tasks 4.x/5.2.
- **Unity runner gotcha**: the aggregated controller runner prints the PREVIOUS passing test's `file:line` as a stale prefix on a FAIL line — attribute failures by test NAME, not the printed file:line prefix. - **Unity runner gotcha**: the aggregated controller runner prints the PREVIOUS passing test's `file:line` as a stale prefix on a FAIL line — attribute failures by test NAME, not the printed file:line prefix.
- **(3.1) Legacy path routes through BARE helpers, not `ha_promote_to_primary`**: `ha_elect_evaluate_and_signal`'s PRIMARY branch calls `ha_register_persist_hooks` + `ha_start_primary_heartbeat` + `ha_arm_own_term_gate` directly (matching the raft path), NOT `ha_promote_to_primary` — because the latter calls `ha_do_promote` (term increment) and the legacy branch historically has NO term store/increment (research §4 finding 2). This is a deliberate deviation from design §283's loose wording ("route through ha_promote_to_primary / ha_register_persist_hooks"); the bare-helper route preserves term semantics and is the correct reading. - **(3.1) Legacy path routes through BARE helpers, not `ha_promote_to_primary`**: `ha_elect_evaluate_and_signal`'s PRIMARY branch calls `ha_register_persist_hooks` + `ha_start_primary_heartbeat` + `ha_arm_own_term_gate` directly (matching the raft path), NOT `ha_promote_to_primary` — because the latter calls `ha_do_promote` (term increment) and the legacy branch historically has NO term store/increment (research §4 finding 2). This is a deliberate deviation from design §283's loose wording ("route through ha_promote_to_primary / ha_register_persist_hooks"); the bare-helper route preserves term semantics and is the correct reading.
- **(4.1) No shims — DELETE, don't stub** (design §Transitional-State Deletions / §Design Synthesis: "rewritten or DELETED, not stubbed"). Machinery-subject test cases (convergence checker, committing watchdog, ack-driven promotion, ack counting) were DELETED, not shimmed: test_ha_role.c (5 majority_convergence + 2 committing_timeout_promotes_degraded), test_ha_rdhup_integration.c (rdhup_committing_to_primary_on_ack, rdhup_3node_one_ack_promotes), test_ha_connection_reuse.c (dispatcher_committing_handles_repl_ack, promoted_ep_repl_ack_promotes_to_primary). Kept state-entry cases had their deleted-fn CALLS excised (not shimmed). `test_ha_connection_reuse.c` was overlooked by the spec's §3 census disposition for machinery cases — confirmed in census (line 97) but its ack cases needed DELETE, not UPDATE.
- **(4.1) Residual for task 6.2 final grep**: 3 DOCUMENTARY PROSE comments in `test_ha_role_lifecycle.c` (~lines 760/815/825) still mention `ha_repl_ack_received`/`committing_timeout_fn` by name (no live code). The 4.1 gate (`src/controller/*.c|*.h`) excludes testsuite so this passed; the task-6.2 repo-wide grep WILL catch them — scrub these comments in 5.x/6.x. Also a non-tracked editor backup `src/controller/.ha.c.~undo-tree~` holds stale refs (irrelevant; not a build input — do not commit).
- **(3.1) Expected-failure set after task 3.1 (8 category-(a), all owned downstream)**: task 4.1 DELETES `test_ha_evaluate_committing.c` (3 fails: test_winning_quorum_enters_committing, test_committing_resets_repl_ack_count, test_election_timeout_fn_fires_evaluate); task 5.1 retargets `test_nonblocking_single_node_election_enters_committing` (test_ha_startup.c), `test_3node_win_calls_both_pushes` (test_ha_repl_push.c), and 3 RDHUP cases (test_ha_rdhup_integration.c: rdhup_full_path_reaches_committing, rdhup_committing_to_primary_on_ack, rdhup_3node_one_ack_promotes). `committing_timeout_fn` now has a benign `-Wunused-function` warning (no arming site left) until task 4.1 deletes it. - **(3.1) Expected-failure set after task 3.1 (8 category-(a), all owned downstream)**: task 4.1 DELETES `test_ha_evaluate_committing.c` (3 fails: test_winning_quorum_enters_committing, test_committing_resets_repl_ack_count, test_election_timeout_fn_fires_evaluate); task 5.1 retargets `test_nonblocking_single_node_election_enters_committing` (test_ha_startup.c), `test_3node_win_calls_both_pushes` (test_ha_repl_push.c), and 3 RDHUP cases (test_ha_rdhup_integration.c: rdhup_full_path_reaches_committing, rdhup_committing_to_primary_on_ack, rdhup_3node_one_ack_promotes). `committing_timeout_fn` now has a benign `-Wunused-function` warning (no arming site left) until task 4.1 deletes it.
- **Lifecycle fixture (test_ha_role_lifecycle.c)**: `lifecycle_cluster_setup` installs `repl_broadcast_send_hook = lc_drop_send_hook` (drop-only, reset to NULL in teardown) — required once promotion broadcasts the WAL_NOOP, else fake `int` endpoints cause an ASAN overflow. Gate cases prime `current_lsn = noop_lsn` so the post-increment append lands the noop at `noop_lsn` (production derives `noop_lsn = current_lsn-1`). - **Lifecycle fixture (test_ha_role_lifecycle.c)**: `lifecycle_cluster_setup` installs `repl_broadcast_send_hook = lc_drop_send_hook` (drop-only, reset to NULL in teardown) — required once promotion broadcasts the WAL_NOOP, else fake `int` endpoints cause an ASAN overflow. Gate cases prime `current_lsn = noop_lsn` so the post-increment append lands the noop at `noop_lsn` (production derives `noop_lsn = current_lsn-1`).

218
src/controller/ha.c
View file

@ -61,13 +61,6 @@ typedef struct {
ha_elect_ctx_t *ctx; ha_elect_ctx_t *ctx;
} election_timeout_ctx_t; } 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 / /* cluster_init_phase2 / push_config_transport_only /
* ha_send_ctrlhello_to_all_standbys: defined in cluster.c / config_repl.c; * ha_send_ctrlhello_to_all_standbys: defined in cluster.c / config_repl.c;
* only referenced in production builds. * 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 void on_vote_connected(endpoint_t *ep);
static int vote_ep_err_handler(void *ep_raw); 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 /* ha-single-leader-role task 2.2: shared promotion seam — every road to
* PRIMARY registers the persist hooks and arms the own-term gate. */ * PRIMARY registers the persist hooks and arms the own-term gate. */
static void ha_register_persist_hooks(cluster_t *c); 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 */ /* Cancel any pending election timeout */
raft_cb_cancel_timeout(ctx); 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 */ /* Sync ha->term */
atomic_store_explicit(&ha->term, term, memory_order_release); 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->client_ssl_ctx = NULL;
ha->phase2_done = 0; ha->phase2_done = 0;
ha->n_peers = 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. /* ha-heartbeat-liveness Task 1.1: zero-init liveness timestamp.
* Written by ha_on_heartbeat_received (task 2.1); read by liveness_poll_fn. */ * 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 * ha_dispatch_repl_action_for_standby — execute the decision-loop action for
* one standby on the current heartbeat cycle (Task 3.3). * 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 */ #endif /* XCE_TESTING */
/* ── Test-only hook shims (XCE_TESTING only) ────────────────────────────── /* ── Test-only hook shims (XCE_TESTING only) ──────────────────────────────
* Type definitions (election_timeout_ctx_t, committing_timeout_ctx_t), * Type definitions (election_timeout_ctx_t), extern declarations
* extern declarations (cluster_init_phase2, push_config_transport_only, * (cluster_init_phase2, push_config_transport_only,
* ha_send_ctrlhello_to_all_standbys), and static forward declarations * ha_send_ctrlhello_to_all_standbys), and static forward declarations
* (committing_timeout_fn, ha_heartbeat_is_fresh) have been moved to the * (ha_heartbeat_is_fresh) have been moved to the forward-declarations block
* forward-declarations block near the top of this file so that the raft * near the top of this file so that the raft callback implementations
* callback implementations (Task 6.2) can reference them. * (Task 6.2) can reference them.
*/ */
#ifdef XCE_TESTING #ifdef XCE_TESTING
@ -2726,7 +2644,7 @@ done:
* repl_ack_decode_and_dispatch: decode an inbound packet on a promoted vote * repl_ack_decode_and_dispatch: decode an inbound packet on a promoted vote
* endpoint while role == CTRL_PRIMARY_COMMITTING or CTRL_PRIMARY. * 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 stateReportMsg → ha_process_standby_state (Task 3.1)
* - OPCode_reply with walackreplymsg → repl_on_ack (Task 4.1) * - OPCode_reply with walackreplymsg → repl_on_ack (Task 4.1)
* - OPCode_reply with replconfigpushreplymsg → repl_on_config_ack * - 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) { if (opcode == OPCode_replCatchUpDone) {
capn_free(&capn_ctx); 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 && } else if (opcode == OPCode_reply &&
outer_rq.body_which == Request_body_replymsg && outer_rq.body_which == Request_body_replymsg &&
outer_rq.body.replymsg.p.data) { 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, sr.config_tag, sr.applied_lsn,
committed_tag, committed_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. /* Derive oldest_wal_lsn from the WAL directory.
* The oldest retained segment's start LSN is the boundary — * The oldest retained segment's start LSN is the boundary —
* standbys below it need a snapshot (Task 3.5b). */ * 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; 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 * repl_catchup_done_pkt_for_test: build a minimal io_packet_t encoding
* Request{opcode=replCatchUpDone, body_which=nodata}. * 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_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 * vote_ep_ctx_t stored at ep->data. Tests that need the promoted-endpoint
* to call ha_promote_to_primary(c) when quorum is reached must call this * dispatch path to reach ha_promote_to_primary(c) with a live cluster pointer
* after vote_ep_ctx_build_for_test (which sets vctx->c = NULL by default). * 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). * Requirements: task 7.3 (ha-peer-connection-reuse).
*/ */
void vote_ep_ctx_set_cluster_for_test(endpoint_t *ep, cluster_t *c) 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 */ #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 ───────────────────────────────────────── /* ── Task 2.4: vote_ep_err_handler ─────────────────────────────────────────
* *
* Called by iomgr when the outgoing vote endpoint encounters an error * Called by iomgr when the outgoing vote endpoint encounters an error

32
src/controller/ha.h
View file

@ -397,26 +397,6 @@ void ha_process_standby_state(ha_state_t *ha, int peer_idx,
uint64_t reported_tag, uint64_t reported_lsn, uint64_t reported_tag, uint64_t reported_lsn,
uint64_t committed_tag, uint64_t committed_lsn); uint64_t committed_tag, uint64_t committed_lsn);
/* ── Task 5.2: Majority-convergence → PRIMARY transition ─────────────────────
*
* ha_check_majority_convergence: count converged standbys and promote to
* CTRL_PRIMARY when the threshold is met while role == CTRL_PRIMARY_COMMITTING.
*
* Threshold: ha->n_peers - 1 (all standbys must converge).
* Single-node (n_peers=1): threshold=0 → satisfied immediately.
*
* Double-promotion is prevented by the single-threaded vtask context; a role
* check (CTRL_PRIMARY_COMMITTING) is sufficient.
* Logs INFO "majority converged (N/M standbys) → CTRL_PRIMARY" (Req 7.4).
* If c != NULL, registers persistence hooks and arms periodic heartbeat.
*
* Called from repl_ack_decode_and_dispatch after ha_process_standby_state.
*
* Requirements: 2.3, 2.4, 6.3, 6.5 (ha-repl-hardening spec)
* Boundary: Controller FSM (ha.c)
*/
void ha_check_majority_convergence(ha_state_t *ha, cluster_t *c);
/* ── Task 3.5c: Standby snapshot receive handler ──────────────────────────── */ /* ── Task 3.5c: Standby snapshot receive handler ──────────────────────────── */
/* /*
@ -628,13 +608,6 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c,
uint64_t my_config_tag, uint64_t my_lsn, uint64_t my_config_tag, uint64_t my_lsn,
const char *my_name); const char *my_name);
/*
* ha_repl_ack_received: called when a standby ACKs config+WAL catch-up
* (OPCode_replCatchUpDone). Increments repl_ack_count; when majority
* threshold is reached, calls ha_promote_to_primary (Req 8.2).
*/
void ha_repl_ack_received(ha_state_t *ha, cluster_t *c);
/* /*
* ha_dispatch_repl_action_for_standby: evaluate the replication decision for * ha_dispatch_repl_action_for_standby: evaluate the replication decision for
* one standby and execute the resulting action. * one standby and execute the resulting action.
@ -793,8 +766,9 @@ extern void (*ha_rdhup_election_hook)(ha_state_t *ha, cluster_t *c,
/* /*
* vote_ep_ctx_set_cluster_for_test: set the cluster_t pointer inside the * vote_ep_ctx_set_cluster_for_test: set the cluster_t pointer inside the
* vote_ep_ctx_t stored at ep->data. Used by unit tests that need * vote_ep_ctx_t stored at ep->data. Used by unit tests that need the
* ha_repl_ack_received to call ha_promote_to_primary(c) when quorum is reached. * promoted-endpoint dispatch path to reach ha_promote_to_primary(c) with a
* live cluster pointer.
* The vote_ep_ctx_t struct is internal to ha.c; this helper avoids exposing it. * The vote_ep_ctx_t struct is internal to ha.c; this helper avoids exposing it.
* Requirements: task 7.3 (ha-peer-connection-reuse). * Requirements: task 7.3 (ha-peer-connection-reuse).
*/ */

2
src/controller/meson.build
View file

@ -115,9 +115,7 @@ controller_unit_test_src = [
'testsuite/test_ha_vote_send.c', 'testsuite/test_ha_vote_send.c',
'testsuite/test_ha_start_election_async.c', 'testsuite/test_ha_start_election_async.c',
'testsuite/test_ha_vote_gate.c', 'testsuite/test_ha_vote_gate.c',
'testsuite/test_ha_evaluate_committing.c',
'testsuite/test_ha_repl_push.c', 'testsuite/test_ha_repl_push.c',
'testsuite/test_ha_repl_ack.c',
'testsuite/test_ha_connection_reuse.c', 'testsuite/test_ha_connection_reuse.c',
'testsuite/test_ha_rdhup_integration.c', 'testsuite/test_ha_rdhup_integration.c',
'testsuite/test_ha_liveness.c', 'testsuite/test_ha_liveness.c',

4
src/controller/testsuite/controller_test.c
View file

@ -45,9 +45,7 @@ extern void test_ha_failover(void);
extern void test_ha_vote_send(void); extern void test_ha_vote_send(void);
extern void test_ha_start_election_async(void); extern void test_ha_start_election_async(void);
extern void test_ha_vote_gate(void); extern void test_ha_vote_gate(void);
extern void test_ha_evaluate_committing(void);
extern void test_ha_repl_push(void); extern void test_ha_repl_push(void);
extern void test_ha_repl_ack(void);
extern void test_ha_connection_reuse(void); extern void test_ha_connection_reuse(void);
extern void test_ha_rdhup_integration(void); extern void test_ha_rdhup_integration(void);
extern void test_ha_liveness(void); extern void test_ha_liveness(void);
@ -168,9 +166,7 @@ int main(int argc, char* argv[])
TEST(test_ha_vote_send); TEST(test_ha_vote_send);
TEST(test_ha_start_election_async); TEST(test_ha_start_election_async);
TEST(test_ha_vote_gate); TEST(test_ha_vote_gate);
TEST(test_ha_evaluate_committing);
TEST(test_ha_repl_push); TEST(test_ha_repl_push);
TEST(test_ha_repl_ack);
TEST(test_ha_connection_reuse); TEST(test_ha_connection_reuse);
TEST(test_ha_rdhup_integration); TEST(test_ha_rdhup_integration);
TEST(test_ha_liveness); TEST(test_ha_liveness);

143
src/controller/testsuite/test_ha_connection_reuse.c
View file

@ -59,7 +59,10 @@ extern endpoint_t *vote_ep_ctx_build_for_test(endpoint_t *ep,
extern void raft_cb_become_leader_for_test(ha_state_t *ha, uint64_t term); extern void raft_cb_become_leader_for_test(ha_state_t *ha, uint64_t term);
extern void raft_cb_become_follower_for_test(ha_state_t *ha, uint64_t term); extern void raft_cb_become_follower_for_test(ha_state_t *ha, uint64_t term);
/* ha_repl_ack_received: public, declared in ha.h */ /* The replication-ack promotion handler was DELETED in production
* (ha-single-leader-role task 4.1: the ack-counting force-promotion path is gone —
* promotion is driven by the heartbeat decision loop + per-entry majority ack).
* The ack-counting cases that depended on it were deleted; no shim. */
/* ha_phase2_hook: suppress cluster_init_phase2 in callbacks */ /* ha_phase2_hook: suppress cluster_init_phase2 in callbacks */
extern int (*ha_phase2_hook)(void *c); extern int (*ha_phase2_hook)(void *c);
@ -173,39 +176,12 @@ static void test_dispatcher_electing_calls_vote_reply_hook(void)
printf("test_dispatcher_electing_calls_vote_reply_hook: PASS\n"); printf("test_dispatcher_electing_calls_vote_reply_hook: PASS\n");
} }
/* ── Test 2: dispatcher routes CTRL_PRIMARY_COMMITTING → repl_ack ────────── */ /* ── Test 2 (dispatcher CTRL_PRIMARY_COMMITTING → repl_ack count) DELETED ─────
* ha-single-leader-role task 4.1. Subject was the deleted ack-counting
static void test_dispatcher_committing_handles_repl_ack(void) * promotion machinery (replCatchUpDone incrementing repl_ack_count to drive
{ * COMMITTING → PRIMARY). That path is gone — nothing left to verify. No shim
ha_state_t ha; * (design §"Transitional-State Deletions").
init_ha_with_standby_eps(&ha); */
ha.n_peers = 2;
atomic_store_explicit(&ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
atomic_store_explicit(&ha.repl_ack_count, 0, memory_order_release);
ha_elect_ctx_t *ctx = alloc_elect_ctx(1);
TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "alloc_elect_ctx must succeed");
endpoint_t ep;
memset(&ep, 0, sizeof(ep));
vote_ep_ctx_build_for_test(&ep, &ha, ctx, 0, 1, 0, 0, "test");
io_packet_t *pkt = repl_catchup_done_pkt_for_test();
TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "repl_catchup_done_pkt_for_test must succeed");
int rc = peer_ctrl_msg_handler_for_test(&ep, pkt);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "handler must return 0");
int acks = atomic_load_explicit(&ha.repl_ack_count, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, acks,
"CTRL_COMMITTING: replCatchUpDone must increment repl_ack_count (Req 1.2)");
free(ep.data);
free_elect_ctx(ctx);
fini_ha_with_standby_eps(&ha);
printf("test_dispatcher_committing_handles_repl_ack: PASS\n");
}
/* ── Test 3: dispatcher drops packet for CTRL_STANDBY ───────────────────── */ /* ── Test 3: dispatcher drops packet for CTRL_STANDBY ───────────────────── */
@ -502,92 +478,14 @@ static void test_follower_passive_no_outbound_connect(void)
printf("test_follower_passive_no_outbound_connect: PASS\n"); printf("test_follower_passive_no_outbound_connect: PASS\n");
} }
/* ── Test 10 (Task 7.3): promoted connection carries WAL replication ACK ──── /* ── Test 10 (Task 7.3: promoted ep repl_ack promotes to PRIMARY) DELETED ─────
* * ha-single-leader-role task 4.1. Subject was the deleted ack-counting
* Req 5.2: OPCode_replCatchUpDone decoded on a promoted endpoint while in * promotion machinery (replCatchUpDone incrementing repl_ack_count until the
* CTRL_PRIMARY_COMMITTING must increment repl_ack_count. When majority * majority threshold drives COMMITTING → CTRL_PRIMARY via ha_promote_to_primary).
* threshold is reached, role transitions to CTRL_PRIMARY via ha_promote_to_primary. * The mechanism is gone — promotion is now the heartbeat decision loop +
* * per-entry majority ack. Nothing left to verify; no shim (design
* Uses a 5-node cluster (n_peers=4: self + 4 peers). * §"Transitional-State Deletions").
* quorum = (4+2)/2 = 3; threshold = quorum-1 = 2.
* So two peer ACKs are required before promotion, allowing us to verify
* the "still COMMITTING after first ACK" invariant.
*
* ha_promote_to_primary requires a non-NULL cluster_t, so we embed the
* ha_state_t inside a zero-init cluster_t and pass &c via
* vote_ep_ctx_set_cluster_for_test. ha_promote_to_primary is safe on a
* zero-init cluster_t: xce_add_hook returns -EINVAL harmlessly (zeroed
* hooksys), and ha_start_primary_heartbeat skips when delaymgr is not running.
*
* Sequence:
* - Set n_peers=4 (5-node cluster → quorum=3 → threshold=2 peer ACKs)
* - Inject first replCatchUpDone → repl_ack_count=1, still COMMITTING
* - Inject second replCatchUpDone → repl_ack_count=2 >= threshold → CTRL_PRIMARY
*/ */
static void test_promoted_ep_repl_ack_promotes_to_primary(void)
{
/* Use cluster_t so ha_promote_to_primary has a valid c->ha to promote */
cluster_t c;
memset(&c, 0, sizeof(c));
ha_init(&c.ha, &g_cfg, 0);
c.ha.standby_endpoints = calloc(1, sizeof(zarray_t));
zarray_init(c.ha.standby_endpoints, 4, NULL);
c.ha.n_peers = 4; /* 5-node: quorum=3, threshold=2 → need 2 peer ACKs */
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
atomic_store_explicit(&c.ha.repl_ack_count, 0, memory_order_release);
ha_elect_ctx_t *ctx = alloc_elect_ctx(1);
TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "alloc_elect_ctx must succeed");
/* First standby ep — wire cluster pointer so promotion can fire */
endpoint_t ep0;
memset(&ep0, 0, sizeof(ep0));
vote_ep_ctx_build_for_test(&ep0, &c.ha, ctx, 0, 1, 0, 0, "test");
vote_ep_ctx_set_cluster_for_test(&ep0, &c);
io_packet_t *pkt0 = repl_catchup_done_pkt_for_test();
TEST_ASSERT_NOT_NULL_MESSAGE(pkt0, "repl_catchup_done_pkt_for_test must succeed");
int rc = peer_ctrl_msg_handler_for_test(&ep0, pkt0);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "first ACK: handler must return 0");
int acks = atomic_load_explicit(&c.ha.repl_ack_count, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, acks,
"Task 7.3: first replCatchUpDone must increment repl_ack_count to 1 (Req 5.2)");
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING,
(int)atomic_load_explicit(&c.ha.role, memory_order_acquire),
"Task 7.3: role must stay COMMITTING after first ACK (quorum not reached)");
/* Second standby ep — same ctx, wire cluster pointer */
endpoint_t ep1;
memset(&ep1, 0, sizeof(ep1));
vote_ep_ctx_build_for_test(&ep1, &c.ha, ctx, 0, 1, 0, 0, "test2");
vote_ep_ctx_set_cluster_for_test(&ep1, &c);
io_packet_t *pkt1 = repl_catchup_done_pkt_for_test();
TEST_ASSERT_NOT_NULL_MESSAGE(pkt1, "second repl_catchup_done_pkt_for_test must succeed");
rc = peer_ctrl_msg_handler_for_test(&ep1, pkt1);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "second ACK: handler must return 0");
acks = atomic_load_explicit(&c.ha.repl_ack_count, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE(2, acks,
"Task 7.3: second replCatchUpDone must increment repl_ack_count to 2 (Req 5.2)");
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"Task 7.3: reaching quorum (2/2) must promote role to CTRL_PRIMARY (Req 5.2)");
free(ep0.data);
free(ep1.data);
free_elect_ctx(ctx);
zarray_fini(c.ha.standby_endpoints);
free(c.ha.standby_endpoints);
c.ha.standby_endpoints = NULL;
ha_fini(&c.ha);
printf("test_promoted_ep_repl_ack_promotes_to_primary: PASS\n");
}
/* ── Test suite entry point ─────────────────────────────────────────────── */ /* ── Test suite entry point ─────────────────────────────────────────────── */
@ -595,7 +493,8 @@ void test_ha_connection_reuse(void)
{ {
/* Req 1.2 — dispatcher routes by role */ /* Req 1.2 — dispatcher routes by role */
RUN_TEST(test_dispatcher_electing_calls_vote_reply_hook); RUN_TEST(test_dispatcher_electing_calls_vote_reply_hook);
RUN_TEST(test_dispatcher_committing_handles_repl_ack); /* test_dispatcher_committing_handles_repl_ack deleted (ha-single-leader-role
* 4.1: ack-counting promotion machinery removed, no shim) */
RUN_TEST(test_dispatcher_standby_drops_packet); RUN_TEST(test_dispatcher_standby_drops_packet);
/* Req 1.3 — winner promotes vote endpoints */ /* Req 1.3 — winner promotes vote endpoints */
@ -614,6 +513,6 @@ void test_ha_connection_reuse(void)
/* Task 7.2 — Req 2.1, 2.4: follower path is fully passive */ /* Task 7.2 — Req 2.1, 2.4: follower path is fully passive */
RUN_TEST(test_follower_passive_no_outbound_connect); RUN_TEST(test_follower_passive_no_outbound_connect);
/* Task 7.3 — Req 5.2: replCatchUpDone on promoted ep promotes to PRIMARY */ /* Task 7.3 (test_promoted_ep_repl_ack_promotes_to_primary) deleted
RUN_TEST(test_promoted_ep_repl_ack_promotes_to_primary); * (ha-single-leader-role 4.1: ack-counting promotion machinery removed, no shim) */
} }

749
src/controller/testsuite/test_ha_evaluate_committing.c
View file

@ -1,749 +0,0 @@
/*
* test_ha_evaluate_committing.c — Unity tests for Task 4.1:
* ha_elect_evaluate_and_signal COMMITTING path.
*
* Tests verify:
* Req 4.1: quorum reached → ha->role transitions to CTRL_PRIMARY_COMMITTING
* (not CTRL_PRIMARY) after ha_elect_evaluate_and_signal
* Req 5.2: COMMITTING path: ha->repl_ack_count reset to 0 when entering COMMITTING
* Req 9.2: cluster_init_phase2 called exactly once on first election;
* second election (RDHUP) skips it (phase2_done=1)
*
* Design ref: design.md "ha_elect_evaluate_and_signal (extended)"
* Boundary: ha.c (election evaluation)
* Requirements: 4.1, 5.2, 9.2
*/
#define _GNU_SOURCE
#include <errno.h>
#include <semaphore.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unity/unity.h>
#include "ha.h"
#include "xce/cluster_def.h" /* full cluster_t layout for committing_timeout_fn tests */
/* ── Test-only seam: cluster_init_phase2 call counter ───────────────────────
*
* In production, ha_elect_evaluate_and_signal calls cluster_init_phase2(c)
* when !ha->phase2_done. For unit tests we cannot provide a real cluster_t,
* so we hook via an XCE_TESTING seam in ha.c: the evaluate function checks
* ha_phase2_hook; if non-NULL it calls the hook instead of cluster_init_phase2.
*
* This avoids pulling in the full cluster_t dependency in the test binary.
*/
extern int (*ha_phase2_hook)(void *c);
/* Call counter for the hook */
static int g_phase2_call_count = 0;
static int g_phase2_hook_rc = 0; /* return value for hook (0 = success) */
static int stub_cluster_init_phase2(void *c)
{
(void)c;
g_phase2_call_count++;
return g_phase2_hook_rc;
}
/* ── Minimal ha_config_t for tests ─────────────────────────────────────── */
static ha_config_t g_cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.election_jitter_max_ms = 0,
.election_priority_step_ms = 500,
.committing_timeout_ms = HA_DEFAULT_COMMITTING_TIMEOUT_MS,
};
/* ── Helper: build a minimal ha_elect_ctx_t for n_other peers ───────────── */
static ha_elect_ctx_t *make_ctx(int n_other, int votes_granted)
{
ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) +
(size_t)n_other * sizeof(ctx->votes[0]));
if (!ctx) return NULL;
ctx->n_peers = n_other;
ctx->votes_granted = votes_granted; /* includes self-vote */
ctx->n_replied = n_other;
ctx->max_resp_tag = 0;
ctx->max_resp_lsn = 0;
ctx->result = -1;
ctx->done_posted = 0;
ctx->c = NULL; /* no real cluster in unit tests */
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
/* Store my_tag/my_lsn in votes[0] for yield gate (only if n_other > 0) */
if (n_other > 0) {
ctx->votes[0].my_tag = 100;
ctx->votes[0].my_lsn = 50;
}
return ctx;
}
static void free_ctx(ha_elect_ctx_t *ctx)
{
if (!ctx) return;
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
free(ctx);
}
/* ── Test setup/teardown ─────────────────────────────────────────────────── */
static void reset_counters(void)
{
g_phase2_call_count = 0;
g_phase2_hook_rc = 0;
/* Wire up the test seam */
ha_phase2_hook = stub_cluster_init_phase2;
}
static void clear_hook(void)
{
ha_phase2_hook = NULL;
}
/* ── Test 1: winning quorum → CTRL_PRIMARY_COMMITTING (not PRIMARY) ─────── */
static void test_winning_quorum_enters_committing(void)
{
/* 3-node cluster: n_other=2, quorum=2. Self-vote + 2 peer grants = 3.
* ha_elect_evaluate_and_signal should detect quorum → COMMITTING. */
reset_counters();
ha_state_t ha;
ha_init(&ha, &g_cfg, 0);
ha.phase2_done = 0; /* first boot */
atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release);
ha_elect_ctx_t *ctx = make_ctx(2, 3); /* votes_granted=3 (self + 2 peers) */
TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "make_ctx failed");
/* Set votes[0] my_tag/my_lsn for yield gate */
ctx->votes[0].my_tag = 100;
ctx->votes[0].my_lsn = 50;
ctrl_role_t result = ha_elect_evaluate_and_signal(&ha, ctx);
/* Function must return CTRL_PRIMARY (the election result) */
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, (int)result,
"ha_elect_evaluate_and_signal must return CTRL_PRIMARY when quorum won");
/* But ha->role must be CTRL_PRIMARY_COMMITTING (Req 4.1) */
ctrl_role_t actual_role =
(ctrl_role_t)atomic_load_explicit(&ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, (int)actual_role,
"ha->role must be CTRL_PRIMARY_COMMITTING after winning election (not CTRL_PRIMARY)");
free_ctx(ctx);
ha_fini(&ha);
clear_hook();
printf("test_winning_quorum_enters_committing: PASS (role=%d==CTRL_PRIMARY_COMMITTING)\n",
(int)actual_role);
}
/* ── Test 2: first boot — phase2 called exactly once ────────────────────── */
static void test_first_boot_phase2_called_once(void)
{
/* phase2_done=0: cluster_init_phase2 must be called exactly once. */
reset_counters();
ha_state_t ha;
ha_init(&ha, &g_cfg, 0);
ha.phase2_done = 0;
atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release);
/* Single-node-like: n_other=0, quorum=1, votes_granted=1 → wins */
ha_elect_ctx_t *ctx = make_ctx(0, 1);
TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "make_ctx failed");
ha_elect_evaluate_and_signal(&ha, ctx);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_phase2_call_count,
"cluster_init_phase2 must be called exactly once on first boot");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, ha.phase2_done,
"ha->phase2_done must be set to 1 after cluster_init_phase2");
free_ctx(ctx);
ha_fini(&ha);
clear_hook();
printf("test_first_boot_phase2_called_once: PASS (calls=%d)\n", g_phase2_call_count);
}
/* ── Test 3: second election (RDHUP) skips phase2 ───────────────────────── */
static void test_rdhup_second_election_skips_phase2(void)
{
/* phase2_done=1: cluster_init_phase2 must NOT be called. */
reset_counters();
ha_state_t ha;
ha_init(&ha, &g_cfg, 0);
ha.phase2_done = 1; /* already initialized (post-RDHUP failover) */
atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release);
ha_elect_ctx_t *ctx = make_ctx(0, 1); /* single-node wins again */
TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "make_ctx failed");
ha_elect_evaluate_and_signal(&ha, ctx);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_phase2_call_count,
"cluster_init_phase2 must NOT be called when phase2_done=1 (RDHUP failover)");
free_ctx(ctx);
ha_fini(&ha);
clear_hook();
printf("test_rdhup_second_election_skips_phase2: PASS (calls=%d)\n", g_phase2_call_count);
}
/* ── Test 4: COMMITTING path resets repl_ack_count ──────────────────────── */
static void test_committing_resets_repl_ack_count(void)
{
/* Req 5.2: repl_ack_count must be 0 when entering COMMITTING.
* Start with a non-zero value to verify it gets reset. */
reset_counters();
ha_state_t ha;
ha_init(&ha, &g_cfg, 0);
ha.phase2_done = 1; /* skip phase2 call */
atomic_store_explicit(&ha.repl_ack_count, 3, memory_order_release); /* stale count */
atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release);
ha_elect_ctx_t *ctx = make_ctx(0, 1); /* wins */
TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "make_ctx failed");
ha_elect_evaluate_and_signal(&ha, ctx);
int ack_count = atomic_load_explicit(&ha.repl_ack_count, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, ack_count,
"ha->repl_ack_count must be reset to 0 when entering CTRL_PRIMARY_COMMITTING");
free_ctx(ctx);
ha_fini(&ha);
clear_hook();
printf("test_committing_resets_repl_ack_count: PASS (repl_ack_count=%d)\n", ack_count);
}
/* ── Test 5: losing election → CTRL_STANDBY, phase2 called on first boot ── */
static void test_losing_election_enters_standby(void)
{
/* n_other=2, quorum=2, votes_granted=1 (self only) → loses.
* On first boot, phase2 must still be called; role → CTRL_STANDBY. */
reset_counters();
ha_state_t ha;
ha_init(&ha, &g_cfg, 0);
ha.phase2_done = 0; /* first boot */
atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release);
ha_elect_ctx_t *ctx = make_ctx(2, 1); /* votes_granted=1, quorum=2 → loses */
TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "make_ctx failed");
ctrl_role_t result = ha_elect_evaluate_and_signal(&ha, ctx);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_STANDBY, (int)result,
"ha_elect_evaluate_and_signal must return CTRL_STANDBY when quorum not won");
ctrl_role_t actual_role =
(ctrl_role_t)atomic_load_explicit(&ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_STANDBY, (int)actual_role,
"ha->role must be CTRL_STANDBY after losing election");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_phase2_call_count,
"cluster_init_phase2 must be called once even for losing standby (first boot)");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, ha.phase2_done,
"ha->phase2_done must be set to 1 after losing first-boot election");
free_ctx(ctx);
ha_fini(&ha);
clear_hook();
printf("test_losing_election_enters_standby: PASS (role=%d==CTRL_STANDBY, phase2=%d)\n",
(int)actual_role, g_phase2_call_count);
}
/* ── Test 6: election timeout fires evaluate → COMMITTING if wins ────────── */
static void test_election_timeout_fn_fires_evaluate(void)
{
/* election_timeout_fn calls ha_elect_evaluate_and_signal.
* Use election_timeout_fn_for_test (XCE_TESTING shim in ha.c):
* election_timeout_fn_for_test(ha, ctx, cancelled=0) → fires evaluate.
* election_timeout_fn_for_test(ha, ctx, cancelled=1) → no-op.
*/
reset_counters();
ha_state_t ha;
ha_init(&ha, &g_cfg, 0);
ha.phase2_done = 1; /* skip phase2 */
atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release);
ha_elect_ctx_t *ctx = make_ctx(0, 1); /* single-node wins */
TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "make_ctx failed");
/* Use the test-shim to simulate the timeout firing with cancelled=0 */
extern int election_timeout_fn_for_test(ha_state_t *ha, ha_elect_ctx_t *ctx,
int cancelled_val);
election_timeout_fn_for_test(&ha, ctx, 0 /* not cancelled */);
ctrl_role_t role =
(ctrl_role_t)atomic_load_explicit(&ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, (int)role,
"election_timeout_fn (not cancelled) must drive ha->role to CTRL_PRIMARY_COMMITTING");
free_ctx(ctx);
ha_fini(&ha);
clear_hook();
printf("test_election_timeout_fn_fires_evaluate: PASS (role=%d)\n", (int)role);
}
/* ── Test 7: cancelled election timeout → no role change ────────────────── */
static void test_cancelled_timeout_is_noop(void)
{
reset_counters();
ha_state_t ha;
ha_init(&ha, &g_cfg, 0);
ha.phase2_done = 1;
atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release);
ha_elect_ctx_t *ctx = make_ctx(0, 1);
TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "make_ctx failed");
extern int election_timeout_fn_for_test(ha_state_t *ha, ha_elect_ctx_t *ctx,
int cancelled_val);
election_timeout_fn_for_test(&ha, ctx, 1 /* cancelled */);
ctrl_role_t role =
(ctrl_role_t)atomic_load_explicit(&ha.role, memory_order_acquire);
/* Role must still be CTRL_ELECTING — cancelled, nothing happened */
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_ELECTING, (int)role,
"cancelled election_timeout_fn must not change ha->role");
free_ctx(ctx);
ha_fini(&ha);
clear_hook();
printf("test_cancelled_timeout_is_noop: PASS (role still CTRL_ELECTING)\n");
}
/* ── Test 8: done_posted guard prevents double-execute ───────────────────── */
static void test_done_posted_prevents_double_execute(void)
{
/* Calling ha_elect_evaluate_and_signal twice on the same ctx must be a no-op
* on the second call (done_posted guard). */
reset_counters();
ha_state_t ha;
ha_init(&ha, &g_cfg, 0);
ha.phase2_done = 0;
atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release);
ha_elect_ctx_t *ctx = make_ctx(0, 1); /* wins */
TEST_ASSERT_NOT_NULL_MESSAGE(ctx, "make_ctx failed");
ha_elect_evaluate_and_signal(&ha, ctx); /* first call: phase2_done→1, role→COMMITTING */
/* Reset phase2_done to detect a second call attempt */
ha.phase2_done = 0;
ha_elect_evaluate_and_signal(&ha, ctx); /* second call: done_posted=1, should no-op */
/* phase2_call_count must still be 1 (second call was no-op) */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_phase2_call_count,
"done_posted guard must prevent cluster_init_phase2 from being called twice");
free_ctx(ctx);
ha_fini(&ha);
clear_hook();
printf("test_done_posted_prevents_double_execute: PASS (calls=%d)\n", g_phase2_call_count);
}
/* ── Test 9: committing_timeout_fn fires ha_promote_to_primary ───────────── */
static void test_committing_timeout_fn_promotes_to_primary(void)
{
/* Req 4.3, 8.4: when not cancelled and role == CTRL_PRIMARY_COMMITTING,
* committing_timeout_fn must call ha_promote_to_primary, which drives
* c->ha.role to CTRL_PRIMARY (degraded promotion). */
reset_counters();
/* Zero-init cluster: xce_add_hook returns -EINVAL harmlessly on zeroed
* hooksys, so ha_promote_to_primary is safe to call (comment at ha.c:603). */
cluster_t c;
memset(&c, 0, sizeof(c));
ha_init(&c.ha, &g_cfg, 0);
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
extern int committing_timeout_fn_for_test(ha_state_t *ha, cluster_t *c,
int cancelled_val);
committing_timeout_fn_for_test(&c.ha, &c, 0 /* not cancelled */);
ctrl_role_t role =
(ctrl_role_t)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, (int)role,
"committing_timeout_fn (not cancelled, COMMITTING role) must promote to CTRL_PRIMARY");
ha_fini(&c.ha);
clear_hook();
printf("test_committing_timeout_fn_promotes_to_primary: PASS (role=%d==CTRL_PRIMARY)\n",
(int)role);
}
/* ── Test 10: cancelled committing_timeout_fn → no role change ───────────── */
static void test_committing_timeout_fn_cancelled_is_noop(void)
{
/* Req 4.3: when cancelled=1, committing_timeout_fn must be a no-op;
* ha->role must remain CTRL_PRIMARY_COMMITTING. */
reset_counters();
cluster_t c;
memset(&c, 0, sizeof(c));
ha_init(&c.ha, &g_cfg, 0);
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
extern int committing_timeout_fn_for_test(ha_state_t *ha, cluster_t *c,
int cancelled_val);
committing_timeout_fn_for_test(&c.ha, &c, 1 /* cancelled */);
ctrl_role_t role =
(ctrl_role_t)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, (int)role,
"cancelled committing_timeout_fn must not change ha->role");
ha_fini(&c.ha);
clear_hook();
printf("test_committing_timeout_fn_cancelled_is_noop: PASS (role still CTRL_PRIMARY_COMMITTING)\n");
}
/* ── Task 4.1 (ha-repl-hardening): ha_compute_safe_committed_lsn ────────────
*
* Computes safe_committed_lsn and safe_config_tag from the vote replies stored
* in ha_elect_ctx_t.votes[] after winning an election.
*
* Algorithm: collect LSNs from all granted voters (including self at votes[0].my_lsn),
* sort descending, take the value at index floor(n_granted/2) (i.e. the majority
* threshold element) — this is the highest LSN held by a majority.
*
* Same logic applies for config tags (peer_tag).
*
* Requirements: 2.3 (ha-repl-hardening spec)
* Design ref: design.md §Flow 0: Post-Election — Determine Safe Commit Point
*/
#ifdef XCE_TESTING
extern void ha_compute_safe_committed_lsn(ha_elect_ctx_t *ctx,
uint64_t *safe_lsn_out,
uint64_t *safe_tag_out);
#endif
/*
* Test case A (from task description):
* 3-node cluster — self at lsn=100, peer0 granted at lsn=99, peer1 granted at lsn=100
* Sorted granted LSNs desc: [100 (self), 100 (peer1), 99 (peer0)]
* Majority threshold for 3 granted = ceil(3/2) = 2; element at index 1 (0-based) = 100
* → safe_committed_lsn = 100
*/
static void test_safe_committed_lsn_two_peers_at_99_and_100(void)
{
#ifdef XCE_TESTING
/* n_other=2: ctx->votes[] has 2 entries */
ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) + 2 * sizeof(ctx->votes[0]));
TEST_ASSERT_NOT_NULL(ctx);
ctx->n_peers = 2; /* 2 other peers */
ctx->votes_granted = 3; /* self + 2 peers granted */
ctx->n_replied = 2;
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
/* Self data: self_lsn/self_tag (task 6.3) and votes[0].my_lsn/my_tag */
ctx->self_lsn = 100;
ctx->self_tag = 5;
ctx->votes[0].my_lsn = 100;
ctx->votes[0].my_tag = 5;
ctx->votes[0].vote_granted = 1; /* peer0: granted, lsn=99 */
ctx->votes[0].peer_lsn = 99;
ctx->votes[0].peer_tag = 5;
ctx->votes[1].vote_granted = 1; /* peer1: granted, lsn=100 */
ctx->votes[1].peer_lsn = 100;
ctx->votes[1].peer_tag = 5;
uint64_t safe_lsn = 0, safe_tag = 0;
ha_compute_safe_committed_lsn(ctx, &safe_lsn, &safe_tag);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(100, safe_lsn,
"voters [99,100] + self=100 → safe_committed_lsn must be 100");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(5, safe_tag,
"voters all at tag=5 → safe_config_tag must be 5");
free_ctx(ctx);
printf("test_safe_committed_lsn_two_peers_at_99_and_100: PASS (safe_lsn=%" PRIu64
", safe_tag=%" PRIu64 ")\n", safe_lsn, safe_tag);
#else
printf("test_safe_committed_lsn_two_peers_at_99_and_100: SKIP (non-test build)\n");
#endif
}
/*
* Test case B (from task description):
* 3-node cluster — self at lsn=100, peer0 granted at lsn=50, peer1 granted at lsn=51
* Sorted granted LSNs desc: [100 (self), 51 (peer1), 50 (peer0)]
* Majority threshold for 3 granted = 2; element at index 1 = 51
* → safe_committed_lsn = 51
*/
static void test_safe_committed_lsn_two_peers_at_50_and_51(void)
{
#ifdef XCE_TESTING
ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) + 2 * sizeof(ctx->votes[0]));
TEST_ASSERT_NOT_NULL(ctx);
ctx->n_peers = 2;
ctx->votes_granted = 3;
ctx->n_replied = 2;
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
ctx->self_lsn = 100;
ctx->self_tag = 5;
ctx->votes[0].my_lsn = 100;
ctx->votes[0].my_tag = 5;
ctx->votes[0].vote_granted = 1; /* peer0: granted, lsn=50 */
ctx->votes[0].peer_lsn = 50;
ctx->votes[0].peer_tag = 5;
ctx->votes[1].vote_granted = 1; /* peer1: granted, lsn=51 */
ctx->votes[1].peer_lsn = 51;
ctx->votes[1].peer_tag = 5;
uint64_t safe_lsn = 0, safe_tag = 0;
ha_compute_safe_committed_lsn(ctx, &safe_lsn, &safe_tag);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(51, safe_lsn,
"voters [50,51] + self=100 → safe_committed_lsn must be 51");
free_ctx(ctx);
printf("test_safe_committed_lsn_two_peers_at_50_and_51: PASS (safe_lsn=%" PRIu64 ")\n",
safe_lsn);
#else
printf("test_safe_committed_lsn_two_peers_at_50_and_51: SKIP (non-test build)\n");
#endif
}
/*
* Test case C: single-node cluster (n_peers=0, no votes[] entries).
* safe_committed_lsn must equal self's own applied_lsn (self is the only majority).
* Req 2.3 / task 6.3: returning 0 would truncate all WAL on re-election — wrong.
* The fix (task 6.3): ha_elect_ctx_t gains self_lsn/self_tag fields populated
* unconditionally by ha_start_election_async; ha_compute_safe_committed_lsn
* returns ctx->self_lsn for the n_peers=0 case.
*/
static void test_safe_committed_lsn_single_node(void)
{
#ifdef XCE_TESTING
ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx));
TEST_ASSERT_NOT_NULL(ctx);
ctx->n_peers = 0;
ctx->votes_granted = 1; /* self-vote only */
ctx->n_replied = 0;
/* Task 6.3: self_lsn/self_tag populated unconditionally */
ctx->self_lsn = 77;
ctx->self_tag = 3;
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
uint64_t safe_lsn = 0xDEAD, safe_tag = 0xDEAD;
ha_compute_safe_committed_lsn(ctx, &safe_lsn, &safe_tag);
/* Single-node: safe_committed_lsn must equal self_lsn (not 0) */
TEST_ASSERT_EQUAL_UINT64_MESSAGE(77, safe_lsn,
"single-node (n_peers=0): safe_committed_lsn must equal self_lsn (77), not 0");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(3, safe_tag,
"single-node (n_peers=0): safe_config_tag must equal self_tag (3), not 0");
free_ctx(ctx);
printf("test_safe_committed_lsn_single_node: PASS (safe_lsn=%" PRIu64 ")\n", safe_lsn);
#else
printf("test_safe_committed_lsn_single_node: SKIP (non-test build)\n");
#endif
}
/*
* Test case D: only one peer but it didn't grant vote.
* Only self vote is in the granted set; n_granted=1, majority threshold index=0.
* safe_lsn should be from the only granted voter (self), but since self's lsn
* is in votes[0].my_lsn, result should reflect that.
* This tests the partial-grant scenario.
*/
static void test_safe_committed_lsn_one_peer_denied(void)
{
#ifdef XCE_TESTING
ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) + 1 * sizeof(ctx->votes[0]));
TEST_ASSERT_NOT_NULL(ctx);
ctx->n_peers = 1; /* 1 other peer */
ctx->votes_granted = 1; /* only self-vote granted */
ctx->n_replied = 1;
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
ctx->self_lsn = 200;
ctx->self_tag = 7;
ctx->votes[0].my_lsn = 200;
ctx->votes[0].my_tag = 7;
ctx->votes[0].vote_granted = 0; /* peer denied vote */
ctx->votes[0].peer_lsn = 150;
ctx->votes[0].peer_tag = 7;
uint64_t safe_lsn = 0, safe_tag = 0;
ha_compute_safe_committed_lsn(ctx, &safe_lsn, &safe_tag);
/* Only self granted (my_lsn=200). n_granted=1, majority index=0 → lsn=200 */
TEST_ASSERT_EQUAL_UINT64_MESSAGE(200, safe_lsn,
"only self granted (my_lsn=200): safe_committed_lsn must be 200");
free_ctx(ctx);
printf("test_safe_committed_lsn_one_peer_denied: PASS (safe_lsn=%" PRIu64 ")\n",
safe_lsn);
#else
printf("test_safe_committed_lsn_one_peer_denied: SKIP (non-test build)\n");
#endif
}
/* ── Task 4.2 (ha-repl-hardening): committed_lsn set to safe_committed_lsn ──
*
* raft_cb_become_leader must set ha->last_known_committed_lsn to the value
* returned by ha_compute_safe_committed_lsn after WAL truncation.
*
* This ensures that in test mode (c == NULL) the decision loop reads the
* correct committed_lsn from ha->last_known_committed_lsn (see decision-loop
* XCE_TESTING path in ha.c).
*
* Requirement 2.3 (ha-repl-hardening): safe_committed_lsn is derived from
* voter replies; entries at or below it survive truncation; entries above are
* removed.
*
* The test uses raft_cb_become_leader_for_test (XCE_TESTING shim) which calls
* raft_cb_become_leader with ha->raft_c = NULL (no real cluster, no WAL I/O).
* The WAL-level truncation behaviour (file entries above safe_lsn deleted) is
* covered by test_persist_truncate.c; this test focuses on the committed_lsn
* field assignment that task 4.2 adds.
*
* Design ref: design.md §Flow 0 step 5: committed_lsn = safe_committed_lsn
*/
#ifdef XCE_TESTING
extern void raft_cb_become_leader_for_test(ha_state_t *ha, uint64_t term);
#endif
/*
* Test E: raft_cb_become_leader sets ha->last_known_committed_lsn = safe_lsn.
*
* 3-node cluster (n_other=2, votes_granted=3):
* votes[0]: my_lsn=200, peer_lsn=100, vote_granted=1
* votes[1]: peer_lsn=150, vote_granted=1
*
* Sorted granted LSNs desc: [200 (self), 150 (peer1), 100 (peer0)]
* Majority threshold for 3 granted = ceil(3/2) = 2; element at index 1 = 150
* → safe_committed_lsn = 150
*
* After raft_cb_become_leader_for_test, ha->last_known_committed_lsn must == 150.
*
* Additionally verify via ha_compute_safe_committed_lsn that entry AT safe_lsn
* (lsn=150) is the boundary: safe_lsn=150 means truncate ABOVE 150 (entries
* 151+ removed); entry at 150 is preserved.
*/
static void test_become_leader_sets_committed_lsn_to_safe_lsn(void)
{
#ifdef XCE_TESTING
reset_counters();
ha_state_t ha;
ha_init(&ha, &g_cfg, 0);
ha.phase2_done = 1; /* skip phase2 in this test */
/* Build elect_ctx with 2 peers; safe_lsn will be 150 */
ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) + 2 * sizeof(ctx->votes[0]));
TEST_ASSERT_NOT_NULL(ctx);
ctx->n_peers = 2;
ctx->votes_granted = 3; /* self + 2 peers */
ctx->n_replied = 2;
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
/* Self data in self_lsn/self_tag (task 6.3) and votes[0].my_lsn/my_tag */
ctx->self_lsn = 200;
ctx->self_tag = 5;
/* votes[0]: self data (my_lsn=200) + peer0 granted at lsn=100 */
ctx->votes[0].my_lsn = 200;
ctx->votes[0].my_tag = 5;
ctx->votes[0].vote_granted = 1;
ctx->votes[0].peer_lsn = 100;
ctx->votes[0].peer_tag = 5;
/* votes[1]: peer1 granted at lsn=150 */
ctx->votes[1].vote_granted = 1;
ctx->votes[1].peer_lsn = 150;
ctx->votes[1].peer_tag = 5;
/* Verify the safe_lsn computation independently */
uint64_t safe_lsn = 0, safe_tag = 0;
ha_compute_safe_committed_lsn(ctx, &safe_lsn, &safe_tag);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(150, safe_lsn,
"ha_compute_safe_committed_lsn: voters [100,150]+self=200 → safe_lsn must be 150");
/* Wire elect_ctx into ha_state so raft_cb_become_leader can read it */
atomic_store_explicit(&ha.elect_ctx, (void *)ctx, memory_order_release);
/* Drive the leader promotion path (c=NULL → no WAL I/O, no phase2) */
raft_cb_become_leader_for_test(&ha, /*term=*/1);
/* Task 4.2: ha->last_known_committed_lsn must equal safe_lsn (150) */
uint64_t committed_lsn =
atomic_load_explicit(&ha.last_known_committed_lsn, memory_order_acquire);
TEST_ASSERT_EQUAL_UINT64_MESSAGE(150, committed_lsn,
"raft_cb_become_leader must set ha->last_known_committed_lsn = safe_lsn (150); "
"WAL entry AT safe_lsn=150 is preserved; entries above 150 are truncated");
sem_destroy(&ctx->done);
pthread_mutex_destroy(&ctx->lock);
free(ctx);
ha_fini(&ha);
clear_hook();
printf("test_become_leader_sets_committed_lsn_to_safe_lsn: PASS "
"(committed_lsn=%" PRIu64 ")\n", committed_lsn);
#else
printf("test_become_leader_sets_committed_lsn_to_safe_lsn: SKIP (non-test build)\n");
#endif
}
/* ── Suite entry point ───────────────────────────────────────────────────── */
void test_ha_evaluate_committing(void)
{
RUN_TEST(test_winning_quorum_enters_committing);
RUN_TEST(test_first_boot_phase2_called_once);
RUN_TEST(test_rdhup_second_election_skips_phase2);
RUN_TEST(test_committing_resets_repl_ack_count);
RUN_TEST(test_losing_election_enters_standby);
RUN_TEST(test_election_timeout_fn_fires_evaluate);
RUN_TEST(test_cancelled_timeout_is_noop);
RUN_TEST(test_done_posted_prevents_double_execute);
RUN_TEST(test_committing_timeout_fn_promotes_to_primary);
RUN_TEST(test_committing_timeout_fn_cancelled_is_noop);
/* Task 4.1 (ha-repl-hardening): safe committed LSN computation */
RUN_TEST(test_safe_committed_lsn_two_peers_at_99_and_100);
RUN_TEST(test_safe_committed_lsn_two_peers_at_50_and_51);
RUN_TEST(test_safe_committed_lsn_single_node);
RUN_TEST(test_safe_committed_lsn_one_peer_denied);
/* Task 4.2 (ha-repl-hardening): committed_lsn set to safe_committed_lsn */
RUN_TEST(test_become_leader_sets_committed_lsn_to_safe_lsn);
}

130
src/controller/testsuite/test_ha_rdhup_integration.c
View file

@ -50,6 +50,14 @@ static int stub_phase2(void *c) { (void)c; g_phase2_calls++; return 0;
static int stub_push_config(void *c) { (void)c; g_push_config_calls++; return 0; } static int stub_push_config(void *c) { (void)c; g_push_config_calls++; return 0; }
static int stub_ctrlhello(void *c) { (void)c; g_ctrlhello_calls++; return 0; } static int stub_ctrlhello(void *c) { (void)c; g_ctrlhello_calls++; return 0; }
/* ha-single-leader-role task 4.1: the replication-ack handler was deleted (the
* ack-counting force-promotion machinery is gone — promotion is driven solely
* by the heartbeat decision loop + per-entry majority ack). No shim is added
* (design §"Transitional-State Deletions": no replacements, no shims). The
* ack-driven-promotion cases were deleted; the surviving state-entry cases no
* longer call it (production ha_elect_evaluate_and_signal now lands the winner
* directly per task 5.1 territory). */
/* ── Shared HA config ───────────────────────────────────────────────────── */ /* ── Shared HA config ───────────────────────────────────────────────────── */
static ha_config_t g_cfg = { static ha_config_t g_cfg = {
@ -196,50 +204,11 @@ static void test_rdhup_full_path_reaches_committing(void)
} }
/* ═══════════════════════════════════════════════════════════════════════════ /* ═══════════════════════════════════════════════════════════════════════════
* Test 2: RDHUP → COMMITTING → PRIMARY via ha_repl_ack_received * Test 2 (RDHUP → PRIMARY via replication-ack handler) DELETED — ha-single-leader-role
* * task 4.1. Subject was the deleted ack-driven force-promotion machinery
* Req 8.2: majority ack in COMMITTING → CTRL_PRIMARY * (the replication-ack handler); nothing left to verify. No shim, no retarget
* Req 6.5: promoted standby serves traffic (dispatch gate opens) * (design §"Transitional-State Deletions").
* ═════════════════════════════════════════════════════════════════════════ */ * ═════════════════════════════════════════════════════════════════════════ */
static void test_rdhup_committing_to_primary_on_ack(void)
{
install_stubs();
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; /* 2-node cluster: quorum=1, threshold=0 */
g_rdhup_state.c = &c;
ha_rdhup_election_hook = rdhup_election_drives_evaluate;
int sv[2];
TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv));
c.ha.primary_ctrl_fd = sv[0];
/* Fire RDHUP → should reach CTRL_PRIMARY_COMMITTING */
ha_on_primary_ctrl_fd_rdhup(&c.ha, &c);
close(sv[1]);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT((int)CTRL_PRIMARY_COMMITTING, role);
/* n_peers=1 → quorum=(1+2)/2=1 → threshold=0 → first ack promotes.
* ha_repl_ack_received increments count; acks(1) >= quorum-1(0) → promote. */
ha_repl_ack_received(&c.ha, &c);
role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"ha_repl_ack_received must drive COMMITTING → CTRL_PRIMARY on quorum ack (Req 8.2)");
clear_stubs();
ha_fini(&c.ha);
printf("test_rdhup_committing_to_primary_on_ack: PASS (role=%d==CTRL_PRIMARY)\n", role);
}
/* ═══════════════════════════════════════════════════════════════════════════ /* ═══════════════════════════════════════════════════════════════════════════
* Test 3: RDHUP in single-node cluster → immediate PRIMARY (no COMMITTING) * Test 3: RDHUP in single-node cluster → immediate PRIMARY (no COMMITTING)
@ -273,15 +242,11 @@ static void test_rdhup_single_node_reaches_primary(void)
ha_on_primary_ctrl_fd_rdhup(&c.ha, &c); ha_on_primary_ctrl_fd_rdhup(&c.ha, &c);
close(sv[1]); close(sv[1]);
/* n_peers=0 → quorum=(0+2)/2=1 → threshold=0 → acks(0) >= 0 → promoted inline */ /* n_peers=0 → ha_elect_evaluate_and_signal single-node shortcircuit lands
* CTRL_PRIMARY directly (no transitional COMMITTING, no ack machinery —
* ha-single-leader-role). The old replication-ack fallback was excised
* (function deleted task 4.1); the PRIMARY assertion stands on its own. */
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire); int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
/* Single-node: no standbys to ack, committing_timeout degrades to PRIMARY.
* The committing timeout is 10s; for single-node we call ha_repl_ack_received
* once to simulate the degraded-mode zero-threshold promotion (threshold=0). */
if (role == (int)CTRL_PRIMARY_COMMITTING) {
ha_repl_ack_received(&c.ha, &c);
role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
}
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role, TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"single-node RDHUP path must reach CTRL_PRIMARY (Req 9.1)"); "single-node RDHUP path must reach CTRL_PRIMARY (Req 9.1)");
@ -291,52 +256,12 @@ static void test_rdhup_single_node_reaches_primary(void)
} }
/* ═══════════════════════════════════════════════════════════════════════════ /* ═══════════════════════════════════════════════════════════════════════════
* Test 4: 3-node cluster RDHUP — two acks needed, premature ack stays COMMITTING * Test 4 (3-node RDHUP — ack-driven promotion) DELETED — ha-single-leader-role
* * task 4.1. Subject was the deleted ack-counting promotion machinery
* Req 8.2: acks < quorum-1 → stays COMMITTING; acks >= quorum-1 → PRIMARY * (the replication-ack handler drives COMMITTING → PRIMARY on threshold acks);
* 3-node (n_peers=2): quorum=2, threshold=1 → need 1 ack to promote * the mechanism is gone, nothing left to verify. No shim (design §"Transitional-
* State Deletions").
* ═════════════════════════════════════════════════════════════════════════ */ * ═════════════════════════════════════════════════════════════════════════ */
static void test_rdhup_3node_one_ack_promotes(void)
{
install_stubs();
cluster_t c;
myconf_t myconf;
make_cluster(&c, &myconf, "ctrl-a");
myconf.peers_cfg.candidates_count = 3;
snprintf(myconf.peers_cfg.candidates[2].name,
sizeof(myconf.peers_cfg.candidates[2].name), "ctrl-c");
myconf.peers_cfg.candidates[2].port = 7003;
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 = 2; /* 3-node: quorum=2, threshold=1 */
g_rdhup_state.c = &c;
ha_rdhup_election_hook = rdhup_election_drives_evaluate;
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);
close(sv[1]);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role,
"3-node RDHUP must enter CTRL_PRIMARY_COMMITTING before acks");
/* First ack: count=1 >= threshold=1 → promotes */
ha_repl_ack_received(&c.ha, &c);
role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"3-node: first ack (count=1 >= threshold=1) must promote to CTRL_PRIMARY (Req 8.2)");
clear_stubs();
ha_fini(&c.ha);
printf("test_rdhup_3node_one_ack_promotes: PASS (role=%d==CTRL_PRIMARY)\n", role);
}
/* ═══════════════════════════════════════════════════════════════════════════ /* ═══════════════════════════════════════════════════════════════════════════
* Test 5: config push and ctrlhello hooks fire during COMMITTING entry * Test 5: config push and ctrlhello hooks fire during COMMITTING entry
@ -422,9 +347,10 @@ static void test_rdhup_primary_dispatch_gate_opens(void)
ha_on_primary_ctrl_fd_rdhup(&c.ha, &c); ha_on_primary_ctrl_fd_rdhup(&c.ha, &c);
close(sv[1]); close(sv[1]);
/* Drive to CTRL_PRIMARY via ack */ /* ha-single-leader-role: ha_elect_evaluate_and_signal lands the winner in
ha_repl_ack_received(&c.ha, &c); * CTRL_PRIMARY directly (no transitional COMMITTING, no ack machinery). The
* old replication-ack drive-step was excised (function deleted task 4.1);
* the dispatch-gate assertion below is the subject of this case. */
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire); int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT((int)CTRL_PRIMARY, role); TEST_ASSERT_EQUAL_INT((int)CTRL_PRIMARY, role);
@ -450,9 +376,11 @@ static void test_rdhup_primary_dispatch_gate_opens(void)
void test_ha_rdhup_integration(void) void test_ha_rdhup_integration(void)
{ {
RUN_TEST(test_rdhup_full_path_reaches_committing); RUN_TEST(test_rdhup_full_path_reaches_committing);
RUN_TEST(test_rdhup_committing_to_primary_on_ack); /* test_rdhup_committing_to_primary_on_ack deleted (ha-single-leader-role 4.1:
* ack-driven promotion machinery removed, no shim) */
RUN_TEST(test_rdhup_single_node_reaches_primary); RUN_TEST(test_rdhup_single_node_reaches_primary);
RUN_TEST(test_rdhup_3node_one_ack_promotes); /* test_rdhup_3node_one_ack_promotes deleted (ha-single-leader-role 4.1:
* ack-counting promotion machinery removed, no shim) */
RUN_TEST(test_rdhup_committing_fires_replication_hooks); RUN_TEST(test_rdhup_committing_fires_replication_hooks);
RUN_TEST(test_rdhup_primary_dispatch_gate_opens); RUN_TEST(test_rdhup_primary_dispatch_gate_opens);
} }

320
src/controller/testsuite/test_ha_repl_ack.c
View file

@ -1,320 +0,0 @@
/* test_ha_repl_ack.c — Unity tests for Task 6.2: ha_repl_ack_received quorum logic.
*
* Tests verify (Req 8.2, 8.3):
* Req 8.2: When acks reach quorum-1 threshold, ha_promote_to_primary is called.
* Req 8.3: CTRL_PRIMARY opens full traffic (dispatch gate PRIMARY case).
*
* Quorum formula: quorum = (n_peers + 2) / 2
* n_peers=1 (solo) → quorum=1 → threshold=0 (quorum-1=0) → first ack promotes
* n_peers=2 (3-node)→ quorum=2 → threshold=1 → first ack promotes
* n_peers=3 (4-node)→ quorum=2 → threshold=1 → first ack promotes
* n_peers=4 (5-node)→ quorum=3 → threshold=2 → second ack promotes (acks >= quorum-1)
*
* Boundary: ha.c unit tests (ha_repl_ack_received, ha_dispatch_gate)
* Requirements: 8.2, 8.3
*/
#define _GNU_SOURCE
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unity/unity.h>
#include "ha.h"
#include "xce/cluster_def.h"
/* ha_repl_ack_received declaration (defined in ha.c) */
void ha_repl_ack_received(ha_state_t *ha, cluster_t *c);
/* ha_dispatch_gate declaration (defined in ha.c) */
int ha_dispatch_gate(ha_state_t *ha, int opcode);
/* OPCode constants needed */
#include "controller.capnp.h"
/* ── Shared HA config ──────────────────────────────────────────────────── */
static ha_config_t g_ack_cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.election_jitter_max_ms = 0,
.election_priority_step_ms = 500,
.committing_timeout_ms = 10000,
};
/* ── Helper: build a minimally-zeroed cluster_t with ha_init applied ─── */
static void init_cluster_committing(cluster_t *c, int n_peers)
{
memset(c, 0, sizeof(*c));
ha_init(&c->ha, &g_ack_cfg, 0);
c->ha.n_peers = n_peers;
/* Simulate being in COMMITTING state (election won, waiting for acks) */
atomic_store_explicit(&c->ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
atomic_store_explicit(&c->ha.repl_ack_count, 0, memory_order_release);
}
/* ── Req 8.2: 2-node cluster (n_peers=2, quorum=2, threshold=1) ──────── */
/* With 2 peers (3 total nodes), quorum=2, threshold=quorum-1=1.
* First ack reaches threshold → promote. */
static void test_repl_ack_2node_first_ack_promotes(void)
{
/* n_peers=2 means 3 nodes total (self + 2 peers).
* quorum = (2+2)/2 = 2; threshold = quorum-1 = 1.
* First ack (count becomes 1 >= 1) → ha_promote_to_primary called. */
cluster_t c;
init_cluster_committing(&c, 2);
ha_repl_ack_received(&c.ha, &c);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"2-node: first ack (count=1 >= threshold=1) must promote to CTRL_PRIMARY (Req 8.2)");
ha_fini(&c.ha);
printf("test_repl_ack_2node_first_ack_promotes: PASS\n");
}
/* ── Req 8.2: 2-node cluster — 0 acks → stays COMMITTING ────────────── */
static void test_repl_ack_2node_no_ack_stays_committing(void)
{
/* With 0 acks, role must remain CTRL_PRIMARY_COMMITTING. */
cluster_t c;
init_cluster_committing(&c, 2);
/* Do NOT call ha_repl_ack_received — verify initial state */
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role,
"2-node: 0 acks → role must remain CTRL_PRIMARY_COMMITTING (Req 8.2)");
ha_fini(&c.ha);
printf("test_repl_ack_2node_no_ack_stays_committing: PASS\n");
}
/* ── Req 8.2: 4-node cluster (n_peers=3, quorum=2, threshold=1) ──────── */
/* quorum = (3+2)/2 = 2; threshold=1. First ack promotes. */
static void test_repl_ack_4node_first_ack_promotes(void)
{
cluster_t c;
init_cluster_committing(&c, 3);
ha_repl_ack_received(&c.ha, &c);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"4-node: first ack (count=1 >= threshold=1) must promote to CTRL_PRIMARY (Req 8.2)");
ha_fini(&c.ha);
printf("test_repl_ack_4node_first_ack_promotes: PASS\n");
}
/* ── Req 8.2: 5-node cluster (n_peers=4, quorum=3, threshold=2) ──────── */
/* quorum = (4+2)/2 = 3; threshold=2.
* First ack (count=1) below threshold → stays COMMITTING.
* Second ack (count=2) reaches threshold → promotes. */
static void test_repl_ack_5node_first_ack_stays_committing(void)
{
cluster_t c;
init_cluster_committing(&c, 4);
ha_repl_ack_received(&c.ha, &c); /* ack count → 1, threshold=2 → no promote */
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role,
"5-node: first ack (count=1 < threshold=2) must stay COMMITTING (Req 8.2)");
ha_fini(&c.ha);
printf("test_repl_ack_5node_first_ack_stays_committing: PASS\n");
}
static void test_repl_ack_5node_second_ack_promotes(void)
{
cluster_t c;
init_cluster_committing(&c, 4);
ha_repl_ack_received(&c.ha, &c); /* ack count → 1 */
/* Reset role back to COMMITTING so we can test from COMMITTING again.
* (In production ha_promote_to_primary isn't called twice; here we need
* to test the threshold on the second call.) */
/* We start clean from count=1 and add one more via a second call. */
ha_fini(&c.ha);
/* Fresh cluster with pre-populated ack_count=1 to test the crossing point */
cluster_t c2;
init_cluster_committing(&c2, 4);
atomic_store_explicit(&c2.ha.repl_ack_count, 1, memory_order_release);
ha_repl_ack_received(&c2.ha, &c2); /* ack count → 2 >= threshold=2 → promote */
int role = (int)atomic_load_explicit(&c2.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"5-node: second ack (count=2 >= threshold=2) must promote to CTRL_PRIMARY (Req 8.2)");
ha_fini(&c2.ha);
printf("test_repl_ack_5node_second_ack_promotes: PASS\n");
}
/* ── Req 8.2: null cluster_t — ha_repl_ack_received must not crash ───── */
/* Design doc: ha_repl_ack_received(ha, c=NULL) increments ack but skips promote. */
static void test_repl_ack_null_cluster_increments_count(void)
{
ha_state_t ha;
ha_init(&ha, &g_ack_cfg, 0);
ha.n_peers = 2;
atomic_store_explicit(&ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
atomic_store_explicit(&ha.repl_ack_count, 0, memory_order_release);
/* c=NULL: must not crash; quorum check still fires but ha_promote_to_primary
* is guarded by `if (c)` in ha_repl_ack_received. */
ha_repl_ack_received(&ha, NULL);
int acks = atomic_load_explicit(&ha.repl_ack_count, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, acks,
"repl_ack with NULL cluster must still increment repl_ack_count (Req 8.2)");
/* Role should NOT have changed to PRIMARY (no cluster to promote via) */
int role = (int)atomic_load_explicit(&ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role,
"repl_ack with NULL cluster must NOT promote role (no cluster_t available)");
ha_fini(&ha);
printf("test_repl_ack_null_cluster_increments_count: PASS\n");
}
/* ── Req 8.3: after promotion, dispatch gate allows full traffic ──────── */
/* Once ha_repl_ack_received fires ha_promote_to_primary, role = CTRL_PRIMARY.
* ha_dispatch_gate in CTRL_PRIMARY state must allow job submission. */
static void test_repl_ack_after_promote_gate_allows_jobsub(void)
{
static int dummy_ep; /* stands in for the ACKing standby's endpoint */
cluster_t c;
init_cluster_committing(&c, 2);
/* FIXTURE FIX (ha-commit-engine task 6.1, bug-411 baseline failure):
* the CTRL_PRIMARY dispatch gate now requires write quorum — for
* n_peers=2 that is >= 1 connected standby. The ack below models a
* standby that connected and ACKed config; in production that standby
* lives in ha->standby_endpoints (raft_cb_become_leader /
* on_standby_connected). Register it so quorum legitimately holds on
* this healthy primary — the gate itself must NOT be weakened. */
c.ha.standby_endpoints = calloc(1, sizeof(zarray_t));
TEST_ASSERT_NOT_NULL(c.ha.standby_endpoints);
zarray_init(c.ha.standby_endpoints, 4, NULL);
zarray_store(c.ha.standby_endpoints, &dummy_ep);
/* Trigger promotion via ack */
ha_repl_ack_received(&c.ha, &c);
/* Verify role is now PRIMARY */
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"after ack promote: role must be CTRL_PRIMARY (Req 8.2)");
/* Dispatch gate in CTRL_PRIMARY must allow OPCode_jobsub (Req 8.3) */
int rc = ha_dispatch_gate(&c.ha, OPCode_jobsub);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"CTRL_PRIMARY dispatch gate must allow OPCode_jobsub (Req 8.3)");
zarray_fini(c.ha.standby_endpoints);
free(c.ha.standby_endpoints);
c.ha.standby_endpoints = NULL;
ha_fini(&c.ha);
printf("test_repl_ack_after_promote_gate_allows_jobsub: PASS\n");
}
/* ── Req 8.3: before promotion, gate still blocks jobsub ─────────────── */
/* In COMMITTING state, before threshold is reached, job ops are blocked. */
static void test_repl_ack_committing_gate_blocks_jobsub_until_promote(void)
{
/* 5-node cluster: threshold=2. First ack leaves us in COMMITTING. */
cluster_t c;
init_cluster_committing(&c, 4);
ha_repl_ack_received(&c.ha, &c); /* count=1, threshold=2 → still COMMITTING */
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role,
"5-node after 1 ack: must remain COMMITTING before threshold (Req 8.2)");
/* Gate must block job submission while still COMMITTING */
int rc = ha_dispatch_gate(&c.ha, OPCode_jobsub);
TEST_ASSERT_NOT_EQUAL_MESSAGE(0, rc,
"COMMITTING gate must block OPCode_jobsub (Req 8.3, 5.1)");
ha_fini(&c.ha);
printf("test_repl_ack_committing_gate_blocks_jobsub_until_promote: PASS\n");
}
/* ── Req 8.2: repl_ack_count accumulates correctly across calls ──────── */
static void test_repl_ack_count_increments_monotonically(void)
{
ha_state_t ha;
ha_init(&ha, &g_ack_cfg, 0);
ha.n_peers = 10; /* large cluster so quorum > 3 calls */
/* n_peers=10 → quorum=(10+2)/2=6 → threshold=5 */
atomic_store_explicit(&ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
atomic_store_explicit(&ha.repl_ack_count, 0, memory_order_release);
ha_repl_ack_received(&ha, NULL); /* count → 1 */
ha_repl_ack_received(&ha, NULL); /* count → 2 */
ha_repl_ack_received(&ha, NULL); /* count → 3 */
int acks = atomic_load_explicit(&ha.repl_ack_count, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE(3, acks,
"repl_ack_count must monotonically increment with each ack call (Req 8.2)");
ha_fini(&ha);
printf("test_repl_ack_count_increments_monotonically: PASS\n");
}
/* ── Req 8.2: null ha pointer — must not crash ────────────────────────── */
static void test_repl_ack_null_ha_is_noop(void)
{
/* ha_repl_ack_received(NULL, ...) must return without crashing */
ha_repl_ack_received(NULL, NULL);
printf("test_repl_ack_null_ha_is_noop: PASS\n");
}
/* ── Test suite entry point ───────────────────────────────────────────── */
void test_ha_repl_ack(void)
{
/* Req 8.2 — quorum threshold: 2-node (3 total nodes) */
RUN_TEST(test_repl_ack_2node_first_ack_promotes);
RUN_TEST(test_repl_ack_2node_no_ack_stays_committing);
/* Req 8.2 — quorum threshold: 4-node (n_peers=3) */
RUN_TEST(test_repl_ack_4node_first_ack_promotes);
/* Req 8.2 — quorum threshold: 5-node (n_peers=4, threshold=2) */
RUN_TEST(test_repl_ack_5node_first_ack_stays_committing);
RUN_TEST(test_repl_ack_5node_second_ack_promotes);
/* Req 8.2 — null cluster guard and ack count accumulation */
RUN_TEST(test_repl_ack_null_cluster_increments_count);
RUN_TEST(test_repl_ack_count_increments_monotonically);
RUN_TEST(test_repl_ack_null_ha_is_noop);
/* Req 8.3 — dispatch gate reflects promotion state */
RUN_TEST(test_repl_ack_after_promote_gate_allows_jobsub);
RUN_TEST(test_repl_ack_committing_gate_blocks_jobsub_until_promote);
}

362
src/controller/testsuite/test_ha_role.c
View file

@ -377,226 +377,16 @@ static void test_process_standby_state_multiple_peers(void)
printf("test_process_standby_state_multiple_peers: PASS\n"); printf("test_process_standby_state_multiple_peers: PASS\n");
} }
/* ── Task 5.2: ha_check_majority_convergence → PRIMARY transition ──────────── /* ── Task 5.2 convergence-checker tests REMOVED (ha-single-leader-role task 4.1)
* *
* After ha_process_standby_state updates standby_converged[], the primary must * The majority-convergence checker was deleted in production (the convergence-
* count the number of converged standbys and promote to CTRL_PRIMARY when the * counting force-promotion path is gone — promotion is driven solely by the
* majority threshold is met and the current role is CTRL_PRIMARY_COMMITTING. * heartbeat decision loop + per-entry majority ack). The machinery-subject
* * cases that exercised it (3node_none / 3node_one / 5node_threshold /
* Threshold formula (Req 2.3): ceil(n_peers/2) - 1 converged standbys — * 3node_all / single_node_immediate) had nothing left to verify and were
* i.e. primary + converged standbys form a majority of the cluster. * deleted per design §"Transitional-State Deletions" (no replacements, no
* Integer form: n_peers / 2 standbys (n=3 → 1, n=5 → 2). * shims). Their RUN_TEST registrations were removed as well.
* Single-node (n_peers<=1): threshold=0 → satisfied immediately.
* n_peers = total cluster size including self.
*
* Requirements: 2.3, 2.4, 6.1, 6.3, 6.5 (ha-repl-hardening)
*/ */
#ifdef XCE_TESTING
void ha_check_majority_convergence(ha_state_t *ha, cluster_t *c);
#endif
/*
* Test: 3-node cluster; no standbys converged (0/2)
* → role must remain CTRL_PRIMARY_COMMITTING.
* Requirements: 2.3, 6.3 (ha-repl-hardening)
*/
static void test_majority_convergence_3node_none_stays_committing(void)
{
#ifdef XCE_TESTING
cluster_t c;
memset(&c, 0, sizeof(c));
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.committing_timeout_ms = HA_DEFAULT_COMMITTING_TIMEOUT_MS,
};
ha_init(&c.ha, &cfg, 0);
c.ha.n_peers = 3; /* 3-node cluster: primary + 2 standbys */
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
/* No standby converged */
c.ha.standby_converged[0] = false;
c.ha.standby_converged[1] = false;
ha_check_majority_convergence(&c.ha, &c);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role,
"3-node: 0/2 standbys converged → must remain CTRL_PRIMARY_COMMITTING (Req 2.3)");
ha_fini(&c.ha);
printf("test_majority_convergence_3node_none_stays_committing: PASS\n");
#else
printf("test_majority_convergence_3node_none_stays_committing: SKIP\n");
#endif
}
/*
* Test: 3-node cluster; 1/2 standbys converged → primary + 1 standby = 2/3
* nodes hold the state = majority → must promote to CTRL_PRIMARY.
* (Requirement 2.3 formula: ceil(3/2)-1 = 1 converged standby suffices.
* A dead minority standby must not force the degraded-mode timeout.)
* Requirements: 2.3, 2.4, 6.3 (ha-repl-hardening)
*/
static void test_majority_convergence_3node_one_standby_promotes(void)
{
#ifdef XCE_TESTING
cluster_t c;
memset(&c, 0, sizeof(c));
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.committing_timeout_ms = HA_DEFAULT_COMMITTING_TIMEOUT_MS,
};
ha_init(&c.ha, &cfg, 0);
c.ha.n_peers = 3;
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
/* Only standby[0] converged; standby[1] not converged */
c.ha.standby_converged[0] = true;
c.ha.standby_converged[1] = false;
ha_check_majority_convergence(&c.ha, &c);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"3-node: 1/2 standbys converged → majority → CTRL_PRIMARY (Req 2.3)");
ha_fini(&c.ha);
printf("test_majority_convergence_3node_one_standby_promotes: PASS\n");
#else
printf("test_majority_convergence_3node_one_standby_promotes: SKIP\n");
#endif
}
/*
* Test: 5-node cluster; threshold = 5/2 = 2 converged standbys.
* 1/4 converged → stays COMMITTING; 2/4 converged → promotes.
* Requirements: 2.3, 6.3 (ha-repl-hardening)
*/
static void test_majority_convergence_5node_threshold(void)
{
#ifdef XCE_TESTING
cluster_t c;
memset(&c, 0, sizeof(c));
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.committing_timeout_ms = HA_DEFAULT_COMMITTING_TIMEOUT_MS,
};
ha_init(&c.ha, &cfg, 0);
c.ha.n_peers = 5;
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
/* 1/4 converged: primary + 1 = 2 of 5 — not a majority → stays */
c.ha.standby_converged[0] = true;
ha_check_majority_convergence(&c.ha, &c);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role,
"5-node: 1/4 standbys converged → must remain COMMITTING (Req 2.3)");
/* 2/4 converged: primary + 2 = 3 of 5 — majority → promotes */
c.ha.standby_converged[1] = true;
ha_check_majority_convergence(&c.ha, &c);
role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"5-node: 2/4 standbys converged → majority → CTRL_PRIMARY (Req 2.3)");
ha_fini(&c.ha);
printf("test_majority_convergence_5node_threshold: PASS\n");
#else
printf("test_majority_convergence_5node_threshold: SKIP\n");
#endif
}
/*
* Test: 3-node cluster (n_peers=2); both standbys converged (2/2)
* → role must transition to CTRL_PRIMARY.
* Requirements: 2.3, 2.4, 6.3 (ha-repl-hardening)
*/
static void test_majority_convergence_3node_all_standbys_promotes(void)
{
#ifdef XCE_TESTING
cluster_t c;
memset(&c, 0, sizeof(c));
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.committing_timeout_ms = HA_DEFAULT_COMMITTING_TIMEOUT_MS,
};
ha_init(&c.ha, &cfg, 0);
c.ha.n_peers = 3; /* 3-node cluster */
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
/* Both standbys converged */
c.ha.standby_converged[0] = true;
c.ha.standby_converged[1] = true;
ha_check_majority_convergence(&c.ha, &c);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"3-node: 2/2 standbys converged → must transition to CTRL_PRIMARY (Req 2.3, 2.4)");
ha_fini(&c.ha);
printf("test_majority_convergence_3node_all_standbys_promotes: PASS\n");
#else
printf("test_majority_convergence_3node_all_standbys_promotes: SKIP\n");
#endif
}
/*
* Test: single-node cluster (n_peers=0)
* → threshold=0, satisfied immediately → must promote to CTRL_PRIMARY.
* Requirements: 6.5, 2.3 (ha-repl-hardening): single-node is always satisfied.
*/
static void test_majority_convergence_single_node_immediate(void)
{
#ifdef XCE_TESTING
cluster_t c;
memset(&c, 0, sizeof(c));
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.committing_timeout_ms = HA_DEFAULT_COMMITTING_TIMEOUT_MS,
};
ha_init(&c.ha, &cfg, 0);
c.ha.n_peers = 1; /* single-node: just self */
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
/* No standbys to converge — threshold=0 must be satisfied immediately */
ha_check_majority_convergence(&c.ha, &c);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"single-node: n_peers=1 → threshold=0 satisfied → must promote to CTRL_PRIMARY (Req 6.5)");
ha_fini(&c.ha);
printf("test_majority_convergence_single_node_immediate: PASS\n");
#else
printf("test_majority_convergence_single_node_immediate: SKIP\n");
#endif
}
/* ── Task 3.3: ha_dispatch_repl_action_for_standby PUSH_CONFIG wiring ────── /* ── Task 3.3: ha_dispatch_repl_action_for_standby PUSH_CONFIG wiring ──────
* *
@ -879,16 +669,16 @@ static void test_evaluate_repl_action_stream_wal_fresh_system(void)
printf("test_evaluate_repl_action_stream_wal_fresh_system: PASS\n"); printf("test_evaluate_repl_action_stream_wal_fresh_system: PASS\n");
} }
/* ── Task 5.3: committing timeout default value and degraded-mode WARN ─────── /* ── Task 5.3: committing timeout default value ──────────────────────────────
* *
* Requirements: 6.4, 7.5 (ha-repl-hardening) * Requirements: 6.4 (ha-repl-hardening)
* *
* Verifies: * The committing_timeout_ms CONFIG knob survives until task 5.2 (it is a
* - HA_DEFAULT_COMMITTING_TIMEOUT_MS == 60000 (safety-net window is 60 s) * different token from the deleted committing-timeout watchdog callback). The
* - committing_timeout_fn_for_test still promotes to CTRL_PRIMARY in degraded * degraded-mode force-promotion cases that drove the watchdog test wrapper
* mode even when only a subset of standbys have converged * were machinery-subject and were deleted (ha-single-leader-role task 4.1,
* - The ha->standby_converged[] state is readable so the WARN log can include * design §"Transitional-State Deletions": no shims). Only the config-default
* the converged/pending counts * assertion remains.
*/ */
/* /*
@ -902,112 +692,6 @@ static void test_committing_timeout_default_is_60s(void)
printf("test_committing_timeout_default_is_60s: PASS\n"); printf("test_committing_timeout_default_is_60s: PASS\n");
} }
/*
* Test: committing_timeout_fn promotes to PRIMARY even when only some standbys
* have converged (degraded mode). The WARN log path fires when timeout expires
* before majority convergence. After promotion, role must be CTRL_PRIMARY.
*
* Requirements: 6.4, 7.5 (ha-repl-hardening)
*/
#ifdef XCE_TESTING
extern int committing_timeout_fn_for_test(ha_state_t *ha, cluster_t *c,
int cancelled_val);
#endif
static void test_committing_timeout_promotes_degraded_with_partial_convergence(void)
{
#ifdef XCE_TESTING
cluster_t c;
memset(&c, 0, sizeof(c));
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.committing_timeout_ms = HA_DEFAULT_COMMITTING_TIMEOUT_MS,
};
ha_init(&c.ha, &cfg, 0);
c.ha.n_peers = 3; /* 3-node cluster; 2 standbys */
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
/* Only 1 of 2 standbys has converged — majority not yet reached */
c.ha.standby_converged[0] = true;
c.ha.standby_converged[1] = false;
/*
* Count converged/pending so the test mirrors what the WARN log reads.
* n_converged=1, n_pending=1 (Req 7.5).
*/
int n_converged = 0;
int n_pending = 0;
for (int i = 0; i < c.ha.n_peers - 1; i++) {
if (c.ha.standby_converged[i])
n_converged++;
else
n_pending++;
}
TEST_ASSERT_EQUAL_INT_MESSAGE(1, n_converged,
"degraded-mode: 1 standby converged before timeout");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, n_pending,
"degraded-mode: 1 standby still pending before timeout");
/* Fire the timeout — must promote despite partial convergence */
committing_timeout_fn_for_test(&c.ha, &c, 0 /* not cancelled */);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"committing timeout: must promote to CTRL_PRIMARY in degraded mode (Req 6.4)");
ha_fini(&c.ha);
printf("test_committing_timeout_promotes_degraded_with_partial_convergence: PASS "
"(n_converged=%d, n_pending=%d)\n", n_converged, n_pending);
#else
printf("test_committing_timeout_promotes_degraded_with_partial_convergence: SKIP\n");
#endif
}
/*
* Test: committing_timeout_fn promotes when NO standbys have converged
* (worst-case degraded mode — all standbys unreachable).
* Requirements: 6.4 (ha-repl-hardening)
*/
static void test_committing_timeout_promotes_degraded_zero_converged(void)
{
#ifdef XCE_TESTING
cluster_t c;
memset(&c, 0, sizeof(c));
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.committing_timeout_ms = HA_DEFAULT_COMMITTING_TIMEOUT_MS,
};
ha_init(&c.ha, &cfg, 0);
c.ha.n_peers = 3; /* 3-node cluster; all standbys unreachable */
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
/* No standbys converged — n_converged=0, n_pending=2 (Req 7.5) */
c.ha.standby_converged[0] = false;
c.ha.standby_converged[1] = false;
committing_timeout_fn_for_test(&c.ha, &c, 0 /* not cancelled */);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"committing timeout: must promote even with 0 converged standbys (degraded, Req 6.4)");
ha_fini(&c.ha);
printf("test_committing_timeout_promotes_degraded_zero_converged: PASS\n");
#else
printf("test_committing_timeout_promotes_degraded_zero_converged: SKIP\n");
#endif
}
/* ── Task 3.3 (ha-wal-replication): dispatch gate allows OPCode_walCatchupEntry /* ── Task 3.3 (ha-wal-replication): dispatch gate allows OPCode_walCatchupEntry
* in CTRL_STANDBY, CTRL_ELECTING, and CTRL_INIT. * in CTRL_STANDBY, CTRL_ELECTING, and CTRL_INIT.
* *
@ -1085,16 +769,12 @@ void test_ha_role(void)
RUN_TEST(test_dispatch_repl_action_no_push_when_converged); RUN_TEST(test_dispatch_repl_action_no_push_when_converged);
/* Task 3.5d: ha_dispatch_repl_action_for_standby SEND_SNAPSHOT wiring */ /* Task 3.5d: ha_dispatch_repl_action_for_standby SEND_SNAPSHOT wiring */
RUN_TEST(test_dispatch_repl_action_send_snapshot); RUN_TEST(test_dispatch_repl_action_send_snapshot);
/* Task 5.2: ha_check_majority_convergence → PRIMARY transition */ /* Task 5.2 convergence-checker cases deleted (ha-single-leader-role 4.1:
RUN_TEST(test_majority_convergence_3node_none_stays_committing); * the majority-convergence checker removed — no shims, machinery-subject) */
RUN_TEST(test_majority_convergence_3node_one_standby_promotes); /* Task 5.3: committing timeout config-default knob (committing_timeout_ms
RUN_TEST(test_majority_convergence_5node_threshold); * survives until task 5.2; the degraded-mode committing-timeout watchdog
RUN_TEST(test_majority_convergence_3node_all_standbys_promotes); * cases were deleted — ha-single-leader-role 4.1, machinery-subject) */
RUN_TEST(test_majority_convergence_single_node_immediate);
/* Task 5.3: committing timeout default value and degraded-mode WARN */
RUN_TEST(test_committing_timeout_default_is_60s); RUN_TEST(test_committing_timeout_default_is_60s);
RUN_TEST(test_committing_timeout_promotes_degraded_with_partial_convergence);
RUN_TEST(test_committing_timeout_promotes_degraded_zero_converged);
/* Task 3.3 (ha-wal-replication): dispatch gate standby-role guard */ /* Task 3.3 (ha-wal-replication): dispatch gate standby-role guard */
RUN_TEST(test_dispatch_gate_allows_wal_catchup_in_standby_roles); RUN_TEST(test_dispatch_gate_allows_wal_catchup_in_standby_roles);
} }