diff --git a/src/controller/ha.c b/src/controller/ha.c index f1b29389..369c559e 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -1753,21 +1753,23 @@ int ha_restore_term(ha_state_t *ha, const char *workdir) /* ── ha_heartbeat_is_fresh (Req 6.2) ──────────────────────────────────────── * - * Returns 1 if the elapsed time since ha->last_heartbeat is strictly less than - * ha->config.heartbeat_timeout_ms (i.e. the primary is still alive). - * Returns 0 if the heartbeat is stale (timeout has expired). + * Returns 1 if the elapsed time since the last primary message + * (ha->last_activity, written by ha_on_heartbeat_received) is strictly less + * than ha->config.heartbeat_timeout_ms (i.e. the primary is still alive). + * Returns 0 if no primary message has been seen yet (last_activity == 0) or + * the timeout has expired. * - * Uses CLOCK_MONOTONIC to avoid wall-clock skew. + * Uses CLOCK_MONOTONIC seconds to avoid wall-clock skew; whole-second + * granularity is sufficient for the 10 s default timeout. */ static int ha_heartbeat_is_fresh(ha_state_t *ha) { - struct timespec now, last; + time_t last = ha->last_activity; /* time_t load is atomic on LP64 */ + if (last == 0) + return 0; /* no primary message seen yet */ + struct timespec now; clock_gettime(CLOCK_MONOTONIC, &now); - pthread_mutex_lock(&ha->lock); - last = ha->last_heartbeat; - pthread_mutex_unlock(&ha->lock); - long ms = (now.tv_sec - last.tv_sec) * 1000L - + (now.tv_nsec - last.tv_nsec) / 1000000L; + long ms = (long)(now.tv_sec - last) * 1000L; return ms < (long)ha->config.heartbeat_timeout_ms; } @@ -1837,8 +1839,9 @@ int handle_request_vote(ha_state_t *ha, memory_order_acquire); if (my_role == CTRL_PRIMARY || my_role == CTRL_PRIMARY_COMMITTING) return 0; /* deny — I am primary (or committing); protect against challengers */ - if (my_role == CTRL_STANDBY && ha_heartbeat_is_fresh(ha)) - return 0; /* deny — my primary is alive */ + if (my_role == CTRL_STANDBY && ha->primary_ctrl_fd >= 0 && + ha_heartbeat_is_fresh(ha)) + return 0; /* deny — my primary is alive (active primary connection) */ /* Req 2.5: catching-up standbys deny all votes */ if (in_catch_up_flag) @@ -1847,9 +1850,10 @@ int handle_request_vote(ha_state_t *ha, /* Protect voted_for check+set from concurrent vote requests on different vtasks */ pthread_mutex_lock(&ha->lock); - /* Req 2.6: already voted for a different candidate in this term → deny */ - if (ha->voted_for_term == term && - ha->voted_for_name[0] != '\0' && + /* Req 2.6: already voted for a DIFFERENT candidate in this term → deny. + * Same-candidate re-grant is idempotent (Raft §5.2: grant iff votedFor ∈ + * {Nil, candidateId}) — matches the duplicate-vote check in raft.c. */ + if (ha->voted_for_term == term && ha->voted_for_name[0] != '\0' && strcmp(ha->voted_for_name, candidate_id) != 0) { pthread_mutex_unlock(&ha->lock); return 0; @@ -2975,6 +2979,49 @@ static int vote_ep_err_handler(void *ep_raw) return 0; } +/* ── ha_init_raft: one-time raft state machine initialization ──────────────── + * + * Called once at startup after ha_restore_term() to wire the raft state + * machine with the correct cluster size and callbacks. Subsequent elections + * use raft_start_election() / raft_on_leader_lost() on this same state. + * + * n_peers: total candidates_count from cluster config (includes self). + */ +void ha_init_raft(ha_state_t *ha, cluster_t *c, int n_peers) +{ + /* Map persisted voted_for_name back to a peer index for initial_voted_for. + * This lets raft_check_and_start_election (RDHUP path) see the prior vote + * without re-reading the file. */ + int restored_voted_for = -1; + if (ha->voted_for_name[0] != '\0' && c && c->myconf) { + const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg; + for (int i = 0; i < pcfg->candidates_count; i++) { + if (strcmp(pcfg->candidates[i].name, ha->voted_for_name) == 0) { + restored_voted_for = i; + break; + } + } + } + + raft_config_t rcfg; + memset(&rcfg, 0, sizeof(rcfg)); + rcfg.n_peers = n_peers; + rcfg.self_idx = ha->my_priority; + rcfg.election_timeout_ms = ha->config.election_timeout_ms; + rcfg.my_priority = ha->my_priority; + rcfg.initial_term = atomic_load_explicit(&ha->term, memory_order_acquire); + rcfg.initial_voted_for = restored_voted_for; + rcfg.cbs = &s_real_cbs; + rcfg.ctx = ha; + raft_init(&ha->raft, &rcfg); + + ha->n_peers = n_peers; + ha->raft_c = c; + INFO("HA: raft initialized (n_peers=%d, self=%d, term=%" PRIu64 ")", + n_peers, ha->my_priority, + atomic_load_explicit(&ha->term, memory_order_acquire)); +} + /* ── Task 6.3: ha_start_election_async — delegate to raft_start_election ──── * * Non-blocking election initiator. Sets up the raft state machine and @@ -3001,45 +3048,8 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c, (void)peers; /* peer connections driven by raft_cb_send_vote_request */ (void)my_name; /* self-vote name stored in on_persist via raft_start_election */ - /* Step 0: wire cluster pointer for raft callbacks. - * Must be set before any raft_* call so callbacks can reach cluster_t. */ - ha->raft_c = c; - - /* Step 0b: re-initialise raft_state_t with the real cluster size and - * the term/voted_for restored from disk by ha_restore_term(). - * - * ha->raft.term may be 0 from ha_init while ha->term was restored to a - * higher value by ha_restore_term() — reconcile here before the election. - * - * self_idx = ha->my_priority (index of this node in candidates[]). - * n_peers = total cluster size (candidates_count); passed directly. - * - * voted_for: map ha->voted_for_name back to a peer index (-1 if none). - */ - { - int restored_voted_for = -1; - if (ha->voted_for_name[0] != '\0' && c && c->myconf) { - const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg; - for (int i = 0; i < pcfg->candidates_count; i++) { - if (strcmp(pcfg->candidates[i].name, ha->voted_for_name) == 0) { - restored_voted_for = i; - break; - } - } - } - - raft_config_t rcfg; - memset(&rcfg, 0, sizeof(rcfg)); - rcfg.n_peers = n_peers; /* total candidates_count */ - rcfg.self_idx = ha->my_priority; - rcfg.election_timeout_ms = ha->config.election_timeout_ms; - rcfg.my_priority = ha->my_priority; - rcfg.initial_term = atomic_load_explicit(&ha->term, memory_order_acquire); - rcfg.initial_voted_for = restored_voted_for; - rcfg.cbs = &s_real_cbs; - rcfg.ctx = ha; - raft_init(&ha->raft, &rcfg); - } + /* Step 0: raft is initialized once at startup by ha_init_raft(); nothing + * to do here. ha->raft_c and ha->n_peers are already set. */ /* Step 1: transition to CTRL_ELECTING before firing RPCs. * raft_start_election fires on_become_leader / on_send_vote_request below; @@ -3255,6 +3265,7 @@ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c) /* Update n_peers tracking — total cluster size (including self) */ ha->n_peers = n_peers; } + } #ifdef XCE_TESTING diff --git a/src/controller/ha.h b/src/controller/ha.h index 99fda909..d95ecd2d 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -151,7 +151,7 @@ typedef struct { _Atomic uint64_t applied_lsn; /* Standby: last applied WAL LSN */ char primary_host[XCE_MAXNAME_LEN]; /* Hostname of current primary */ uint32_t primary_port; /* Port of current primary */ - struct timespec last_heartbeat; /* CLOCK_MONOTONIC timestamp of last heartbeat */ + /* last_heartbeat deleted — heartbeat freshness now derives from last_activity */ _Atomic int promotion_failed; _Atomic int promotion_retry_count; _Atomic int in_catch_up; /* Non-zero while processing historical WAL entries */ @@ -159,7 +159,7 @@ typedef struct { int hooks_registered; /* Idempotency guard for xce_add_hook calls */ uint32_t last_received_crc; /* Chained CRC of last received replicated entry */ ha_config_t config; - pthread_mutex_t lock; /* Protects primary_host/port + last_heartbeat */ + pthread_mutex_t lock; /* Protects primary_host/port + voted_for_term/name */ /* watchdog_stop / watchdog_thread deleted — replaced by ha_arm_heartbeat_timeout_vtask */ void *ssl_ctx; /* SSL_CTX* cast; avoids pulling openssl headers here */ uint64_t jitter_seed; /* xorshift64 PRNG state for election jitter */ @@ -441,6 +441,21 @@ int ha_handle_snapshot_chunk(ha_state_t *ha, persist_t *p, */ void ha_persist_term(ha_state_t *ha, const char *workdir); +/* + * ha_init_raft: initialize the embedded raft state machine with cluster topology. + * + * Must be called ONCE at controller startup after config is loaded and + * ha_restore_term() has run. Subsequent elections (initial and re-elections + * after primary loss) reuse the same raft state — only raft_start_election() + * and raft_on_leader_lost() are called for each election cycle. + * + * Calling raft_init() more than once would reset raft's internal state + * (term, voted_for, role) and could lose durable election progress. + * + * n_peers: total cluster size including self (candidates_count from config). + */ +void ha_init_raft(ha_state_t *ha, cluster_t *c, int n_peers); + /* * ha_restore_term: read {workdir}/ha-term.bin at startup and restore * ha->term and ha->voted_for_* into ha_state_t. diff --git a/src/controller/myapp.c b/src/controller/myapp.c index 6ff65cf6..5d7dda2a 100644 --- a/src/controller/myapp.c +++ b/src/controller/myapp.c @@ -217,6 +217,11 @@ int myapp_run(myconf_t *myconf) uint64_t my_lsn = (uint64_t)atomic_load_explicit(&c.ha.applied_lsn, memory_order_acquire); + /* Initialize raft state machine once with cluster topology. + * Must happen before any election or watchdog path that calls + * raft_on_leader_lost() / raft_start_election(). */ + ha_init_raft(&c.ha, &c, n_peers); + /* Startup stagger: delay election start by priority * election_priority_step_ms. * Node 0 starts first; lower-priority nodes wait, see the winner's vote request * arrive, and grant the vote instead of starting their own election. @@ -240,11 +245,24 @@ int myapp_run(myconf_t *myconf) * path (iomgr was already running); the raft term + primary_ctrl_fd checks * catch the case where iomgr was NOT running during the nap (connections * queued in the kernel backlog) but the winner's ctrlhello has already - * established the control channel. */ + * established the control channel. + * + * voted_for_name is combined with term: a self-vote means this node was + * a candidate, not a follower — no external leader is known. In + * production raft_cb_persist records the self-vote under this node's own + * name (myconf->name); the "self" placeholder only appears in test mode, + * so both spellings must be checked. A stale vote (voted_for_term < + * ha->term) is also discarded. All these cases must fall through to + * ha_start_election_async. */ { int leader_known = 0; pthread_mutex_lock(&c.ha.lock); - if (c.ha.voted_for_name[0] != '\0') + uint64_t vft = c.ha.voted_for_term; + uint64_t cur = atomic_load_explicit(&c.ha.term, memory_order_relaxed); + if (c.ha.voted_for_name[0] != '\0' && + strcmp(c.ha.voted_for_name, "self") != 0 && + strcmp(c.ha.voted_for_name, myconf->name) != 0 && + vft == cur) leader_known = 1; pthread_mutex_unlock(&c.ha.lock); if (!leader_known && c.ha.raft.term > 0) @@ -253,10 +271,10 @@ int myapp_run(myconf_t *myconf) leader_known = 1; if (leader_known) { INFO("myapp_run: leader already known after startup delay " - "(voted_for=%s, raft_term=%" PRIu64 ", primary_fd=%d) — " - "skipping own election", + "(voted_for=%s, voted_term=%" PRIu64 ", raft_term=%" PRIu64 + ", primary_fd=%d) — skipping own election", c.ha.voted_for_name[0] ? c.ha.voted_for_name : "(none)", - c.ha.raft.term, c.ha.primary_ctrl_fd); + vft, c.ha.raft.term, c.ha.primary_ctrl_fd); goto skip_election; } } @@ -281,6 +299,11 @@ skip_election:; atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY, memory_order_release); INFO("myapp_run: CTRL_STANDBY — waiting for primary to connect and push config"); + /* Arm heartbeat watchdog even when we skipped the election. + * If the known primary never connects within heartbeat_timeout_ms, + * the watchdog fires ha_on_primary_ctrl_fd_rdhup which increments + * the term (clearing voted_for) and starts a fresh election. */ + ha_arm_heartbeat_timeout_vtask(&c.ha, &c); } } diff --git a/src/controller/testsuite/test_ha_vote_gate.c b/src/controller/testsuite/test_ha_vote_gate.c index 78cc1ded..5e7768f0 100644 --- a/src/controller/testsuite/test_ha_vote_gate.c +++ b/src/controller/testsuite/test_ha_vote_gate.c @@ -47,24 +47,16 @@ static ha_config_t g_gate_cfg = { .election_priority_step_ms = 500, }; -/* ── Helper: set ha->last_heartbeat to "now - offset_ms milliseconds ago" ── */ +/* ── Helper: set ha->last_activity to "now - offset_ms milliseconds ago" ── + * + * last_activity is the production field written by ha_on_heartbeat_received + * (whole CLOCK_MONOTONIC seconds); offset_ms is rounded down to seconds. */ -static void set_last_heartbeat_age(ha_state_t *ha, long offset_ms) +static void set_last_activity_age(ha_state_t *ha, long offset_ms) { struct timespec now; clock_gettime(CLOCK_MONOTONIC, &now); - - long sec = offset_ms / 1000; - long nsec = (offset_ms % 1000) * 1000000L; - - ha->last_heartbeat.tv_sec = now.tv_sec - sec; - ha->last_heartbeat.tv_nsec = now.tv_nsec - nsec; - - /* Carry / borrow normalisation */ - if (ha->last_heartbeat.tv_nsec < 0) { - ha->last_heartbeat.tv_sec -= 1; - ha->last_heartbeat.tv_nsec += 1000000000L; - } + ha->last_activity = now.tv_sec - offset_ms / 1000; } /* ── Req 6.1: CTRL_PRIMARY → deny regardless of candidate freshness ────── */ @@ -139,8 +131,13 @@ static void test_standby_fresh_heartbeat_denies_vote_request(void) atomic_store_explicit(&ha.term, 2ULL, memory_order_release); ha.voted_for_term = 0; - /* last_heartbeat = 500 ms ago (< 10 000 ms timeout → fresh) */ - set_last_heartbeat_age(&ha, 500L); + /* Active primary control channel (handle_ctrlhello sets this on a real + * standby; the deny gate requires a live connection AND a fresh heartbeat). + * ha_fini does not close the fd, so any non-negative value is safe. */ + ha.primary_ctrl_fd = 999; + + /* last_activity = 500 ms ago → rounds to "this second" (< 10 s → fresh) */ + set_last_activity_age(&ha, 500L); int granted = handle_request_vote(&ha, 2ULL, @@ -174,8 +171,11 @@ static void test_standby_stale_heartbeat_evaluates_vote(void) atomic_store_explicit(&ha.term, 1ULL, memory_order_release); ha.voted_for_term = 0; - /* last_heartbeat = 15 000 ms ago (> 10 000 ms timeout → stale) */ - set_last_heartbeat_age(&ha, 15000L); + /* Active primary control channel — staleness alone must open the gate */ + ha.primary_ctrl_fd = 999; + + /* last_activity = 15 000 ms ago (> 10 000 ms timeout → stale) */ + set_last_activity_age(&ha, 15000L); int granted = handle_request_vote(&ha, 1ULL, diff --git a/src/lib/raft/raft.c b/src/lib/raft/raft.c index 37b0de7c..e05f9b3f 100644 --- a/src/lib/raft/raft.c +++ b/src/lib/raft/raft.c @@ -40,7 +40,11 @@ int raft_init(raft_state_t *s, const raft_config_t *cfg) s->term = cfg->initial_term; s->role = RAFT_FOLLOWER; s->voted_for = cfg->initial_voted_for; - s->voted_for_term = 0; + /* A restored vote belongs to the restored term: persistence writes + * (term, voted_for) as a pair. Leaving voted_for_term at 0 would let the + * duplicate-vote check miss after restart, granting a second vote in the + * same term (Election Safety violation). */ + s->voted_for_term = (cfg->initial_voted_for != -1) ? cfg->initial_term : 0; s->n_peers = cfg->n_peers; s->self_idx = cfg->self_idx; s->votes_granted = 0; @@ -132,22 +136,35 @@ int raft_on_vote_request(raft_state_t *s, int from_idx, uint64_t term, /* reply_term reflects our (possibly updated) current term from here on */ *reply_term = s->term; - /* Step 3: role gate — LEADER denies */ + /* Step 3: role gate — same-term challenger denied by live leader. + * + * This only fires when term == s->term (the equal-term case). For + * term > s->term, Step 2 already downgraded s->role to RAFT_FOLLOWER + * before we reach here, so the check below is false and we fall through + * to grant/deny based on freshness. + * + * A RequestVote for the same term as the current leader means the sender + * started a competing candidacy in a term the leader already won — deny. */ if (s->role == RAFT_LEADER) { if (s->cbs->on_trace) s->cbs->on_trace(s->ctx, "vote_denied_role", NULL); return RAFT_DENY; } - /* Step 4: duplicate-vote deny — already voted for a *different* candidate - * this same term. If voted_for == from_idx and voted_for_term == term, - * fall through to idempotent re-grant path below. */ - if (s->voted_for != -1 && - s->voted_for != from_idx && - s->voted_for_term == term) { - if (s->cbs->on_trace) - s->cbs->on_trace(s->ctx, "vote_denied_already_voted", NULL); - return RAFT_DENY; + /* Step 4: duplicate-vote check — already voted in this term. + * Idempotent re-grant: if this is the same candidate we already voted for + * in this term, fall through so we re-persist and re-grant (network retry + * path). This is safe: the candidate has already committed to this term + * and we already durably recorded the vote. + * Different candidate: deny — standard Raft invariant of at most one vote + * per node per term. */ + if (s->voted_for != -1 && s->voted_for_term == term) { + if (s->voted_for != from_idx) { + if (s->cbs->on_trace) + s->cbs->on_trace(s->ctx, "vote_denied_already_voted", NULL); + return RAFT_DENY; + } + /* same candidate — fall through to re-persist and re-grant */ } /* Step 5: application-level policy gate (freshness, log lsn, etc.) */ diff --git a/src/lib/testsuite/test_raft.c b/src/lib/testsuite/test_raft.c index e29f78d6..12faf0fe 100644 --- a/src/lib/testsuite/test_raft.c +++ b/src/lib/testsuite/test_raft.c @@ -2029,6 +2029,9 @@ static void test_proto_H1_init_with_persisted_state(void) TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_UINT64(5, s.term); TEST_ASSERT_EQUAL_INT(1, s.voted_for); + /* The restored vote belongs to the restored term — without this, the + * duplicate-vote check cannot match after restart (see H6). */ + TEST_ASSERT_EQUAL_UINT64(5, s.voted_for_term); TEST_ASSERT_EQUAL_INT(RAFT_FOLLOWER, s.role); } @@ -2061,6 +2064,51 @@ static void test_proto_H2_init_first_boot(void) TEST_ASSERT_EQUAL_INT(RAFT_FOLLOWER, s.role); } +/* + * test_proto_H6_restored_vote_blocks_second_vote + * + * Election Safety across restart: a node that voted for candidate 1 in + * term 5, then restarted (raft_init with initial_term=5, initial_voted_for=1), + * must DENY a vote request from a DIFFERENT candidate (idx=2) in the same + * term 5. Without restoring voted_for_term alongside voted_for, the + * duplicate-vote check cannot match the restored term and the node would + * grant a second vote in term 5 → two leaders in one term. + */ +static void test_proto_H6_restored_vote_blocks_second_vote(void) +{ + raft_state_t s; + raft_callbacks_t cbs = make_mock_cbs(); + uint64_t reply_term = 0; + + reset_mock(); + + raft_config_t cfg; + memset(&s, 0, sizeof(s)); + memset(&cfg, 0, sizeof(cfg)); + cfg.n_peers = 3; + cfg.self_idx = 0; + cfg.election_timeout_ms = 1000; + cfg.initial_term = 5; + cfg.initial_voted_for = 1; /* persisted vote for candidate 1, term 5 */ + cfg.cbs = &cbs; + + TEST_ASSERT_EQUAL_INT(0, raft_init(&s, &cfg)); + + /* Different candidate (idx=2) requests a vote in the SAME term 5 */ + int rc = raft_on_vote_request(&s, 2, 5, &reply_term); + + TEST_ASSERT_EQUAL_INT(RAFT_DENY, rc); + TEST_ASSERT_EQUAL_UINT64(5, reply_term); + + /* The restored vote must be untouched */ + TEST_ASSERT_EQUAL_INT(1, s.voted_for); + TEST_ASSERT_EQUAL_UINT64(5, s.voted_for_term); + + /* Same candidate (idx=1) retransmitting must still be re-granted */ + rc = raft_on_vote_request(&s, 1, 5, &reply_term); + TEST_ASSERT_EQUAL_INT(RAFT_GRANT, rc); +} + /* * test_proto_H3_persist_before_vote_recorded * @@ -2964,6 +3012,7 @@ void test_raft(void) RUN_TEST(test_proto_H3_persist_before_vote_recorded); RUN_TEST(test_proto_H4_persist_fail_denies); RUN_TEST(test_proto_H5_start_election_self_vote_persisted); + RUN_TEST(test_proto_H6_restored_vote_blocks_second_vote); RUN_TEST(test_proto_Hnull_required_callback_einval); /* Group D: Quorum and Result (task 5.3) */ diff --git a/src/tools/deploy-ha.sh b/src/tools/deploy-ha.sh index 01917287..b446d304 100755 --- a/src/tools/deploy-ha.sh +++ b/src/tools/deploy-ha.sh @@ -52,12 +52,13 @@ if [ -z "\$NAME" ]; then echo "Usage: \$0 " exit 1 fi +shift CTRL_DIR=$TOP/controller/\$NAME if [ ! -d "\$CTRL_DIR/conf" ] || [ ! -d "\$CTRL_DIR/data" ]; then echo "Error: \$CTRL_DIR/conf or \$CTRL_DIR/data does not exist" exit 1 fi -$BINARY --myname \$NAME --peers "$PEERS_STR" --workdir $TOP/controller --certdir $TOP/certs --logfile $TOP/logs/\$NAME.log --vcpus 4 \$VERBOSE +$BINARY --myname \$NAME --peers "$PEERS_STR" --workdir $TOP/controller --certdir $TOP/certs --logfile $TOP/logs/\$NAME.log --vcpus 4 \$* EOF chmod +x "$TOP/scripts/start.sh"