diff --git a/.kiro/specs/election-iomgr-refactor/tasks.md b/.kiro/specs/election-iomgr-refactor/tasks.md index 2df27258..9f3f645f 100644 --- a/.kiro/specs/election-iomgr-refactor/tasks.md +++ b/.kiro/specs/election-iomgr-refactor/tasks.md @@ -58,14 +58,14 @@ - _Depends: 1.1, 1.2, 2.2, 2.3_ - [ ] 3. Core: FSM gate extensions and role-based vote denial -- [ ] 3.1 (P) Extend ha_dispatch_gate for CTRL_PRIMARY_COMMITTING state +- [x] 3.1 (P) Extend ha_dispatch_gate for CTRL_PRIMARY_COMMITTING state - Add `case CTRL_PRIMARY_COMMITTING:` to `ha_dispatch_gate` in ha.c: allow `OPCode_heartbeat`, `OPCode_replConfig`, `OPCode_replConfigPush`, `OPCode_upgradeConfigAck`, `OPCode_hello`, `OPCode_walAck`, `OPCode_replEntry`, `OPCode_replAck`, `OPCode_replCatchUpDone`; deny all others with `-XCE_ERR_NOT_READY` - Observable: unit test sends `OPCode_jobsubmit` in COMMITTING state → returns `XCE_ERR_NOT_READY`; `OPCode_replCatchUpDone` → returns 0 - _Requirements: 5.1, 5.2, 8.1_ - _Boundary: ha.c (dispatch gate)_ - _Depends: 2.1_ -- [ ] 3.2 (P) Extend handle_request_vote with role-based deny gate +- [x] 3.2 (P) Extend handle_request_vote with role-based deny gate - Before existing term/voted_for/freshness checks in `handle_request_vote`: load `ha->role`; return 0 (deny) if `CTRL_PRIMARY` or `CTRL_PRIMARY_COMMITTING`; return 0 (deny) if `CTRL_STANDBY && ha_heartbeat_is_fresh(ha)` - Implement `ha_heartbeat_is_fresh(ha)`: compare `clock_gettime(CLOCK_MONOTONIC)` against `ha->last_heartbeat`; return 1 if elapsed < `heartbeat_timeout_ms` - Observable: unit test with mocked roles: PRIMARY→deny, COMMITTING→deny, STANDBY+fresh→deny, STANDBY+stale→evaluate, ELECTING→evaluate @@ -74,7 +74,7 @@ - _Depends: 2.1_ - [ ] 4. Core: committing phase — replication push and availability gate -- [ ] 4.1 Extend ha_elect_evaluate_and_signal with COMMITTING path +- [x] 4.1 Extend ha_elect_evaluate_and_signal with COMMITTING path - Extend `ha_elect_evaluate_and_signal`: after determining role=PRIMARY, check `!ha->phase2_done`; if first boot call `cluster_init_phase2(c)` then set `ha->phase2_done=1`; set `ha->role = CTRL_PRIMARY_COMMITTING`; reset `repl_ack_count=0`; schedule committing timeout watchdog via `delay_task_ms` - On STANDBY result: if first boot call `cluster_init_phase2(c)` + set `phase2_done=1`; set role=CTRL_STANDBY; election timeout callback sets `cancelled=1` - Observable: unit test verifies `cluster_init_phase2` called exactly once on first election; second election (RDHUP) skips it; role enters COMMITTING not PRIMARY after winning @@ -90,7 +90,7 @@ - _Boundary: ha.c + cluster.c (replication push)_ - _Depends: 4.1_ -- [ ] 4.3 Implement ha_repl_ack_received and majority-ack → PRIMARY transition +- [x] 4.3 Implement ha_repl_ack_received and majority-ack → PRIMARY transition - Implement `ha_repl_ack_received(ha, c)`: atomically increment `repl_ack_count`; compute quorum `= (ha->n_peers + 2) / 2 - 1`; if `repl_ack_count >= quorum`: set committing timeout `cancelled=1`, call `ha_promote_to_primary(c)` (sets role=CTRL_PRIMARY, enables hooks) - Wire `ha_repl_ack_received` into `OPCode_replCatchUpDone` handler in cluster.c - Observable: unit test: 2 of 3 acks arrive → PRIMARY; 1 of 3 acks → stays COMMITTING @@ -98,7 +98,7 @@ - _Boundary: ha.c + cluster.c (ack counting)_ - _Depends: 4.1_ -- [ ] 4.4 Implement committing timeout watchdog (degraded-mode fallback) +- [x] 4.4 Implement committing timeout watchdog (degraded-mode fallback) - Define `committing_timeout_ctx_t`: `_Atomic int cancelled`, `ha_state_t *ha`, `cluster_t *c` - Implement `committing_timeout_fn`: if `!cancelled` and `ha->role == CTRL_PRIMARY_COMMITTING`, log WARN "committing timeout — opening in degraded mode", call `ha_promote_to_primary(c)` - Schedule via `delay_task_ms(delaymgr, committing_timeout_ms, 1, committing_timeout_fn, ctx)` in the COMMITTING entry path diff --git a/src/controller/ha.c b/src/controller/ha.c index f5218c07..0b323e93 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -927,6 +927,46 @@ int handle_request_vote(ha_state_t *ha, return 1; } +/* ── Type definitions needed by ha_elect_evaluate_and_signal (Task 4.1) ───── + * + * election_timeout_ctx_t and committing_timeout_ctx_t are defined here (before + * ha_elect_evaluate_and_signal) because the COMMITTING path references them. + * vote_ep_ctx_t is defined in the iomgr vote-send section below. + */ + +typedef struct { + _Atomic int cancelled; + ha_state_t *ha; + cluster_t *c; + ha_elect_ctx_t *ctx; +} election_timeout_ctx_t; + +/* committing_timeout_ctx_t: owns the heap allocation; freed by the callback. */ +typedef struct { + _Atomic int cancelled; + ha_state_t *ha; + cluster_t *c; +} committing_timeout_ctx_t; + +/* committing_timeout_fn: forward declaration — defined after ha_elect_evaluate_and_signal */ +static int committing_timeout_fn(void *arg); + +/* cluster_init_phase2: defined in cluster.c. Only referenced in production builds + * (XCE_TESTING uses ha_phase2_hook instead, and cluster.c is not linked into the + * unit-test binary). */ +#ifndef XCE_TESTING +extern int cluster_init_phase2(cluster_t *c); +#endif + +#ifdef XCE_TESTING +/* + * ha_phase2_hook: test-only seam. + * When non-NULL, ha_elect_evaluate_and_signal() calls this INSTEAD OF the real + * cluster_init_phase2() so tests can count phase2 calls without a live cluster_t. + */ +int (*ha_phase2_hook)(void *c) = NULL; +#endif /* XCE_TESTING */ + /* ── Election Evaluation (Req 2.1, 2.7, 2.8) ─────────────────────────────── * * ha_elect_evaluate_and_signal: evaluate accumulated votes; set ctx->result @@ -947,14 +987,14 @@ int handle_request_vote(ha_state_t *ha, */ ctrl_role_t ha_elect_evaluate_and_signal(ha_state_t *ha, ha_elect_ctx_t *ctx) { - (void)ha; /* ha->term used for fencing in network path; not needed here */ - /* Guard: called from handle_request_vote_reply under ctx->lock; called from * unit-test path (peers==NULL) without threads — done_posted prevents double - * sem_post in the network path if two threads both hit quorum simultaneously. */ + * execution if two threads both hit quorum simultaneously. */ if (ctx->done_posted) return (ctx->result == 1) ? CTRL_PRIMARY : CTRL_STANDBY; + ctrl_role_t role_result; + /* Fix 2 (CRITICAL): zero-peer guard — single-node shortcircuit. * n_peers==0 means no other nodes; votes[] flexible array has 0 entries, * so accessing votes[0] would be out-of-bounds. Self-vote (votes_granted=1) @@ -963,49 +1003,156 @@ ctrl_role_t ha_elect_evaluate_and_signal(ha_state_t *ha, ha_elect_ctx_t *ctx) ctx->result = 1; ctx->done_posted = 1; sem_post(&ctx->done); - return CTRL_PRIMARY; + role_result = CTRL_PRIMARY; + goto apply_role; } - int total_nodes = ctx->n_peers + 1; /* include self */ - int quorum = (total_nodes + 1) / 2; /* ceil(total/2) */ + { + int total_nodes = ctx->n_peers + 1; /* include self */ + int quorum = (total_nodes + 1) / 2; /* ceil(total/2) */ - /* Req 2.8: fatal gate — candidate has no config but a peer does */ - uint64_t my_tag = ctx->votes[0].my_tag; - uint64_t my_lsn = ctx->votes[0].my_lsn; - if (my_tag == 0 && ctx->max_resp_tag > 0) { - ctx->result = 0; + /* Req 2.8: fatal gate — candidate has no config but a peer does */ + uint64_t my_tag = ctx->votes[0].my_tag; + uint64_t my_lsn = ctx->votes[0].my_lsn; + if (my_tag == 0 && ctx->max_resp_tag > 0) { + ctx->result = 0; + ctx->done_posted = 1; + sem_post(&ctx->done); + role_result = CTRL_STANDBY; + goto apply_role; + } + + /* Check quorum */ + if (ctx->votes_granted < quorum) { + ctx->result = 0; + ctx->done_posted = 1; + sem_post(&ctx->done); + role_result = CTRL_STANDBY; + goto apply_role; + } + + /* Req 2.7: yield gate — fresher responding peer */ + if (ctx->max_resp_tag > my_tag) { + ctx->result = 0; + ctx->done_posted = 1; + sem_post(&ctx->done); + role_result = CTRL_STANDBY; + goto apply_role; + } + if (ctx->max_resp_tag == my_tag && ctx->max_resp_lsn > my_lsn) { + ctx->result = 0; + ctx->done_posted = 1; + sem_post(&ctx->done); + role_result = CTRL_STANDBY; + goto apply_role; + } + + /* Quorum met and no fresher peer → PRIMARY */ + ctx->result = 1; ctx->done_posted = 1; sem_post(&ctx->done); - return CTRL_STANDBY; + role_result = CTRL_PRIMARY; } - /* Check quorum */ - if (ctx->votes_granted < quorum) { - ctx->result = 0; - ctx->done_posted = 1; - sem_post(&ctx->done); - return CTRL_STANDBY; +apply_role: + /* ── Task 4.1: apply role and call cluster_init_phase2 ───────────────── + * + * After determining the election result, update ha->role and (on first boot) + * call cluster_init_phase2. This runs once per process lifetime — the + * phase2_done flag prevents re-entry on subsequent elections (RDHUP failover). + * + * The cluster_t pointer is carried in ctx->c (NULL in unit tests). + * + * Cancel the election timeout watchdog if we won (it would be a no-op now + * but the explicit cancel avoids a spurious second evaluate call). + */ + { + cluster_t *c = ctx->c; + int is_first_boot = !ha->phase2_done; + + if (role_result == CTRL_PRIMARY) { + /* Cancel election timeout watchdog (quorum already reached) */ + if (ctx->timeout_ctx) { + election_timeout_ctx_t *etc = + (election_timeout_ctx_t *)ctx->timeout_ctx; + atomic_store(&etc->cancelled, 1); + } + + /* First boot: call cluster_init_phase2 to load config+data (Req 9.2) */ + if (is_first_boot) { +#ifdef XCE_TESTING + /* In test mode, use hook if set; skip real phase2 (cluster.c not linked) */ + if (ha_phase2_hook != NULL) + ha_phase2_hook(c); +#else + if (c != NULL) + cluster_init_phase2(c); +#endif + ha->phase2_done = 1; + } + + /* Enter COMMITTING: block job traffic until majority ack (Req 4.1, 8.1) */ + atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY_COMMITTING, + memory_order_release); + INFO("HA FSM: → CTRL_PRIMARY_COMMITTING (term=%" PRIu64 + ", first_boot=%d)", atomic_load_explicit(&ha->term, memory_order_acquire), + is_first_boot); + + /* Reset ack counter for this committing round (Req 5.2) */ + atomic_store(&ha->repl_ack_count, 0); + + /* Arm committing timeout watchdog (degraded-mode fallback, Req 8.4) */ + if (c) { + committing_timeout_ctx_t *ct = calloc(1, sizeof(*ct)); + if (ct) { + atomic_init(&ct->cancelled, 0); + ct->ha = ha; + ct->c = c; + int rc = delay_task_ms(&c->delaymgr, + (int)ha->config.committing_timeout_ms, + 1, committing_timeout_fn, ct); + if (rc < 0) { + WARN("ha_elect_evaluate_and_signal: delay_task_ms " + "for committing timeout failed: %d", rc); + free(ct); + } + } + } + } else { + /* STANDBY result */ + + /* Cancel election timeout watchdog — it is now a no-op but explicit + * cancellation matches spec wording and avoids a spurious lock-acquire + * if the timeout fires after STANDBY is decided. */ + if (ctx->timeout_ctx) { + election_timeout_ctx_t *etc = + (election_timeout_ctx_t *)ctx->timeout_ctx; + atomic_store(&etc->cancelled, 1); + } + + /* First boot: call cluster_init_phase2 so structures are ready; + * config arrives via primary's replication push (Req 9.2, 7.3) */ + if (is_first_boot) { +#ifdef XCE_TESTING + /* In test mode, use hook if set; skip real phase2 (cluster.c not linked) */ + if (ha_phase2_hook != NULL) + ha_phase2_hook(c); +#else + if (c != NULL) + cluster_init_phase2(c); +#endif + ha->phase2_done = 1; + } + + atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY, + memory_order_release); + INFO("HA FSM: → CTRL_STANDBY (term=%" PRIu64 ", first_boot=%d)", + atomic_load_explicit(&ha->term, memory_order_acquire), + is_first_boot); + } } - /* Req 2.7: yield gate — fresher responding peer */ - if (ctx->max_resp_tag > my_tag) { - ctx->result = 0; - ctx->done_posted = 1; - sem_post(&ctx->done); - return CTRL_STANDBY; - } - if (ctx->max_resp_tag == my_tag && ctx->max_resp_lsn > my_lsn) { - ctx->result = 0; - ctx->done_posted = 1; - sem_post(&ctx->done); - return CTRL_STANDBY; - } - - /* Quorum met and no fresher peer → PRIMARY */ - ctx->result = 1; - ctx->done_posted = 1; - sem_post(&ctx->done); - return CTRL_PRIMARY; + return role_result; } /* ── Vote Reply Handler (Req 2.2–2.6) ────────────────────────────────────── @@ -1086,10 +1233,6 @@ void handle_request_vote_reply(ha_state_t *ha, * vote_ep_ctx_t: per-peer context stored in ep->data for each outgoing vote * endpoint. on_vote_connected fires after the TLS handshake completes (when * endpoint_t.ssl_handshake_done transitions to 1 in io_handler). - * - * election_timeout_ctx_t: argument for the delay_task_ms watchdog that fires - * ha_elect_evaluate_and_signal when the election deadline expires without all - * peers replying. Uses an _Atomic int cancelled flag for lock-free cancellation. */ typedef struct { @@ -1103,13 +1246,6 @@ typedef struct { const char *my_name; } vote_ep_ctx_t; -typedef struct { - _Atomic int cancelled; - ha_state_t *ha; - cluster_t *c; - ha_elect_ctx_t *ctx; -} election_timeout_ctx_t; - /* xce_request: defined in request.c, not yet in any public header */ extern int xce_request(endpoint_t *ep, xce_request_t *req); @@ -1483,6 +1619,51 @@ static int election_timeout_fn(void *arg) return 0; } +/* ── Task 4.1: committing_timeout_fn — delaymgr callback ─────────────────── + * + * Fires ha_promote_to_primary (degraded mode) if the majority ack threshold + * has not been reached within committing_timeout_ms. Req 8.4. + * 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) { + WARN("HA: committing timeout — majority ack not received within timeout; " + "promoting to PRIMARY (degraded mode, Req 8.4)"); + if (t->c) + ha_promote_to_primary(t->c); + } + free(t); + return 0; +} + +/* ── Task 4.1: ha_repl_ack_received ──────────────────────────────────────── + * + * Called when a standby ACKs config+WAL catch-up (OPCode_replCatchUpDone). + * Increments repl_ack_count; when majority threshold (quorum-1 peers, since + * primary counts itself) is reached, disarms the committing timeout and calls + * ha_promote_to_primary. Req 8.2. + */ +void ha_repl_ack_received(ha_state_t *ha, cluster_t *c) +{ + if (!ha) return; + int acks = atomic_fetch_add(&ha->repl_ack_count, 1) + 1; + int quorum = (ha->n_peers + 2) / 2; /* majority of total cluster */ + if (acks >= quorum - 1) { /* -1: primary counts itself */ + INFO("HA: majority ack received (acks=%d, quorum=%d) → CTRL_PRIMARY", + acks, quorum); + if (c) + ha_promote_to_primary(c); + } +} + /* ── Task 2.4: vote_ep_err_handler ───────────────────────────────────────── * * Called by iomgr when the outgoing vote endpoint encounters an error @@ -1558,28 +1739,23 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c, ctx->max_resp_lsn = 0; ctx->result = -1; ctx->done_posted = 0; + ctx->c = c; /* Task 4.1: pass cluster for phase2 + committing timeout */ pthread_mutex_init(&ctx->lock, NULL); sem_init(&ctx->done, 0, 0); - /* Store candidate own tag/lsn in votes[0] for evaluate function. - * n_peers==1 → n_other==0, so votes[] has 0 entries; use ctx fields. */ + /* Store candidate own tag/lsn in ctx fields (votes[] has 0 entries + * for n_peers==0 shortcircuit — votes[0] would be out-of-bounds). */ ctx->max_resp_tag = my_config_tag; ctx->max_resp_lsn = my_lsn; - /* Ensure votes[0] fields are accessible (ctx has 0 flexible entries; - * votes[0] would be out-of-bounds — use a separate approach for the - * evaluate function's my_tag/my_lsn reads). */ /* For single-node, quorum = ceil(1/2) = 1; votes_granted=1 → wins. - * Fix 1 (CRITICAL): enter CTRL_PRIMARY_COMMITTING, not CTRL_PRIMARY. - * With no standbys the committing timeout (task 4.4) will fire and - * promote to CTRL_PRIMARY (degraded/single-node mode, Req 8.4). */ + * Task 4.1: ha_elect_evaluate_and_signal now handles COMMITTING entry, + * phase2, repl_ack_count reset, and committing timeout arming. */ INFO("HA election: outcome=PRIMARY term=%" PRIu64 " votes_granted=1 quorum=1 (single-node async)", cur_term); - INFO("HA FSM: CTRL_ELECTING → CTRL_PRIMARY_COMMITTING " - "(single-node async, term=%" PRIu64 ")", cur_term); - atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY_COMMITTING, - memory_order_release); + /* evaluate sets ha->role = CTRL_PRIMARY_COMMITTING and calls phase2 */ + ha_elect_evaluate_and_signal(ha, ctx); pthread_mutex_destroy(&ctx->lock); sem_destroy(&ctx->done); @@ -1604,6 +1780,7 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c, ctx->max_resp_lsn = 0; ctx->result = -1; ctx->done_posted = 0; + ctx->c = c; /* Task 4.1: pass cluster for phase2 + committing timeout */ pthread_mutex_init(&ctx->lock, NULL); sem_init(&ctx->done, 0, 0); @@ -2053,6 +2230,7 @@ ctrl_role_t am_i_primary_quorum(ha_state_t *ha, ctx->max_resp_lsn = 0; ctx->result = -1; ctx->done_posted = 0; + ctx->c = NULL; /* am_i_primary_quorum has no cluster_t */ pthread_mutex_init(&ctx->lock, NULL); sem_init(&ctx->done, 0, 0); diff --git a/src/controller/ha.h b/src/controller/ha.h index cd2ff806..f7d0c1e4 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -96,6 +96,10 @@ typedef struct { * (quorum-reached path) can set timeout_ctx->cancelled = 1 to cancel the * watchdog. Set by ha_start_election_async after scheduling the timeout. */ void *timeout_ctx; /* election_timeout_ctx_t*; NULL until watchdog scheduled */ + /* Task 4.1: cluster_t* so ha_elect_evaluate_and_signal can call + * cluster_init_phase2(c) and arm the committing timeout watchdog. + * NULL in unit tests (c=NULL bypasses phase2 and watchdog arming). */ + cluster_t *c; ha_vote_entry_t votes[]; /* C99 flexible array: n_peers-1 entries */ } ha_elect_ctx_t; @@ -496,9 +500,24 @@ int ha_find_and_connect_to_primary(ha_state_t *ha, #define HA_FIND_PRIMARY_T3_INIT_MS 100 #define HA_FIND_PRIMARY_T3_MAX_MS 5000 +/* + * 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); + /* ── P3-6: Test-only failover shim ────────────────────────────── */ #ifdef XCE_TESTING +/* + * ha_phase2_hook: test-only seam for cluster_init_phase2. + * When non-NULL, ha_elect_evaluate_and_signal() calls this INSTEAD OF the + * real cluster_init_phase2() so unit tests can count phase2 calls without + * requiring a real cluster_t. Pattern mirrors xce_vote_request_hook. + */ +extern int (*ha_phase2_hook)(void *c); + /* Test-only shim: demote then re-promote; returns new term. * Exercises the manual failover state machine without a live cluster_t. * Calls ha_do_demote() then ha_do_promote(); validates role transitions diff --git a/src/controller/meson.build b/src/controller/meson.build index 224caf60..48c46089 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -114,6 +114,7 @@ controller_unit_test_src = [ 'testsuite/test_ha_vote_send.c', 'testsuite/test_ha_start_election_async.c', 'testsuite/test_ha_vote_gate.c', + 'testsuite/test_ha_evaluate_committing.c', 'startup_validation.c', 'worker.c', 'persist.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 8d3a2a89..e5156ed2 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -46,6 +46,7 @@ extern void test_ha_failover(void); extern void test_ha_vote_send(void); extern void test_ha_start_election_async(void); extern void test_ha_vote_gate(void); +extern void test_ha_evaluate_committing(void); void setUp(void) { return; } @@ -161,6 +162,7 @@ int main(int argc, char* argv[]) TEST(test_ha_vote_send); TEST(test_ha_start_election_async); TEST(test_ha_vote_gate); + TEST(test_ha_evaluate_committing); return UNITY_END(); } diff --git a/src/controller/testsuite/test_ha_evaluate_committing.c b/src/controller/testsuite/test_ha_evaluate_committing.c new file mode 100644 index 00000000..fd55ac3a --- /dev/null +++ b/src/controller/testsuite/test_ha_evaluate_committing.c @@ -0,0 +1,381 @@ +/* + * 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 +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "ha.h" + +/* ── 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 = 10000, +}; + +/* ── 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); +} + +/* ── 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); +}