diff --git a/.kiro/specs/raft-library/tasks.md b/.kiro/specs/raft-library/tasks.md index e874710..a53ae2e 100644 --- a/.kiro/specs/raft-library/tasks.md +++ b/.kiro/specs/raft-library/tasks.md @@ -147,7 +147,7 @@ - _Requirements: 9.1, 9.2_ - _Boundary: ha.h, ha.c adapter init_ -- [ ] 6.2 Implement all raft_callbacks_t callbacks in ha.c +- [x] 6.2 Implement all raft_callbacks_t callbacks in ha.c - `on_send_vote_request`: encode capnp `VoteRequest` and send via existing `endpoint_t`/iomgr path - `on_become_leader`: invoke existing ha leadership promotion path (phase2 trigger) - `on_become_follower`: invoke existing ha standby transition path diff --git a/src/controller/ha.c b/src/controller/ha.c index 75f6291..c035c8e 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -35,25 +35,486 @@ int XCE_ERR_NOT_READY = E_XCE_NOT_READY; int XCE_ERR_STANDBY = E_XCE_STANDBY; -/* ── No-op raft callback stubs (used by ha_init) ───────────── */ +/* No-op stubs removed: s_real_cbs (below) is used exclusively since task 6.2. */ -static int stub_send_vote(void *c, int p, uint64_t t) { (void)c;(void)p;(void)t; return 0; } -static void stub_become_leader(void *c, uint64_t t) { (void)c;(void)t; } -static void stub_become_follower(void *c, uint64_t t, int l) { (void)c;(void)t;(void)l; } -static int stub_persist(void *c, uint64_t t, int v) { (void)c;(void)t;(void)v; return 0; } -static void stub_sched_timeout(void *c, uint32_t d) { (void)c;(void)d; } -static void stub_cancel_timeout(void *c) { (void)c; } -static void stub_start_after_delay(void *c, uint32_t d) { (void)c;(void)d; } +/* ── Forward declarations needed by real raft callbacks (Task 6.2) ──────── + * These static functions and types are defined later in this file; declared + * here so the callback implementations can reference them. */ -static const raft_callbacks_t s_noop_cbs = { - .on_send_vote_request = stub_send_vote, - .on_vote_eligible = NULL, - .on_become_leader = stub_become_leader, - .on_become_follower = stub_become_follower, - .on_persist = stub_persist, - .on_schedule_timeout = stub_sched_timeout, - .on_cancel_timeout = stub_cancel_timeout, - .on_start_election_after_delay = stub_start_after_delay, +/* vote_ep_ctx_t: per-peer context for outgoing vote endpoints (full def below) */ +typedef struct { + ha_state_t *ha; + cluster_t *c; + ha_elect_ctx_t *ctx; + int peer_idx; + uint64_t my_term; + uint64_t my_tag; + uint64_t my_lsn; + const char *my_name; +} vote_ep_ctx_t; + +/* election_timeout_ctx_t: delaymgr callback context for raft election timeout */ +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; + +/* cluster_init_phase2 / push_config_to_standbys / ha_send_ctrlhello_to_all_standbys: + * defined in cluster.c / config_repl.c; only referenced in production builds. + * (XCE_TESTING uses hook shims and does not link cluster.c.) */ +#ifndef XCE_TESTING +extern int cluster_init_phase2(cluster_t *c); +extern void ha_send_ctrlhello_to_all_standbys(cluster_t *c); +extern int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, + const uint8_t sha256[32], uint64_t new_tag); +#endif + +static void on_vote_connected(endpoint_t *ep); +static int vote_ep_err_handler(void *ep_raw); +static int committing_timeout_fn(void *arg); +static int ha_heartbeat_is_fresh(ha_state_t *ha); + +/* ── Real raft callback implementations (Task 6.2) ───────────── + * + * These are registered as s_real_cbs in ha_init, replacing the no-op stubs. + * They are wired into the active election path in task 6.3. + * + * All callbacks receive ctx = ha_state_t*. + * ha->raft_c is set by ha_start_election_async before any raft_* calls. + * When ha->raft_c is NULL (unit test mode), callbacks that require cluster_t + * are no-ops. + */ + +/* + * raft_cb_send_vote_request: encode and send a capnp VoteRequest to peer. + * + * peer_idx is the absolute candidate index in c->myconf->peers_cfg.candidates[]. + * Uses the existing iomgr vote-send path (on_vote_connected / vote_ep_ctx_t). + * Reads the current election context from ha->elect_ctx (set before raft_start_election). + * Returns 0 on success; non-zero if no cluster context or alloc failure. + */ +static int raft_cb_send_vote_request(void *ctx, int peer_idx, uint64_t term) +{ + ha_state_t *ha = (ha_state_t *)ctx; + cluster_t *c = ha->raft_c; + if (!c || !c->myconf) return 1; /* test mode */ + + const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg; + if (peer_idx < 0 || peer_idx >= pcfg->candidates_count) + return 1; + + ha_elect_ctx_t *ectx = (ha_elect_ctx_t *)atomic_load_explicit( + &ha->elect_ctx, memory_order_acquire); + if (!ectx) return 1; /* no election in progress */ + + xce_candidate_t *peer = (xce_candidate_t *)&pcfg->candidates[peer_idx]; + + /* Compute the local index of peer_idx within ectx->votes[]: + * votes[] has n_peers - 1 = n_other entries for peers excluding self. + * Map: votes_idx = peer_idx if peer_idx < self_idx, else peer_idx - 1. */ + int self_idx = ha->my_priority; /* self_idx == my_priority by convention */ + int votes_idx = (peer_idx < self_idx) ? peer_idx : peer_idx - 1; + if (votes_idx < 0 || votes_idx >= ectx->n_peers) + return 1; + + /* Allocate per-peer context */ + vote_ep_ctx_t *vctx = calloc(1, sizeof(*vctx)); + if (!vctx) { + handle_request_vote_reply(ha, ectx, votes_idx, term, 0, 0, 0); + return 1; + } + vctx->ha = ha; + vctx->c = c; + vctx->ctx = ectx; + vctx->peer_idx = votes_idx; + vctx->my_term = term; + /* tag and lsn come from votes[0] which holds our own values */ + vctx->my_tag = (ectx->n_peers > 0) ? ectx->votes[0].my_tag : 0; + vctx->my_lsn = (ectx->n_peers > 0) ? ectx->votes[0].my_lsn : 0; + vctx->my_name = c->myconf ? c->myconf->name : ""; + + /* Get a fresh endpoint */ + endpoint_t *ep = endpoint_get(); + if (!ep) { + free(vctx); + handle_request_vote_reply(ha, ectx, votes_idx, term, 0, 0, 0); + return 1; + } + + endpoint_init(ep, AF_INET, SOCK_STREAM, 0, 0); + if (c) endpoint_set_peer(ep, c->myhost); + ep->data = vctx; + ep->iomgr = &c->iomgr; + + if (ha->client_ssl_ctx) { + SSL_CTX *client_ctx = (SSL_CTX *)ha->client_ssl_ctx; + ep->ssl = SSL_new(client_ctx); + if (ep->ssl) + SSL_set_connect_state(ep->ssl); + } + + endpoint_set_on_connected(ep, on_vote_connected); + endpoint_set_err_handler(ep, vote_ep_err_handler); + + int rc = endpoint_connect(ep, peer->hostname, peer->port); + if (rc < 0) { + endpoint_fini(ep); + if (!c) + handle_request_vote_reply(ha, ectx, votes_idx, term, 0, 0, 0); + return 1; + } + + rc = iomgr_monitor(ep); + if (rc < 0) { + endpoint_fini(ep); /* frees ep->data (vctx) via FREEUP */ + handle_request_vote_reply(ha, ectx, votes_idx, term, 0, 0, 0); + return 1; + } + + return 0; +} + +/* + * raft_cb_vote_eligible: freshness gate delegated from raft library. + * + * from_idx is the absolute candidate index of the requester. + * Applies the same checks as handle_request_vote: + * - role gate: PRIMARY or COMMITTING → deny + * - fresh heartbeat on STANDBY → deny + * - in_catch_up → deny + * - freshness: candidate config_tag/lsn must be ≥ ours + * Returns 1 to grant, 0 to deny. + */ +static int raft_cb_vote_eligible(void *ctx, int from_idx, uint64_t term) +{ + (void)from_idx; (void)term; + ha_state_t *ha = (ha_state_t *)ctx; + + /* Role gate */ + ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role, + memory_order_acquire); + if (role == CTRL_PRIMARY || role == CTRL_PRIMARY_COMMITTING) + return 0; + + /* Fresh heartbeat on STANDBY → primary is alive, deny */ + if (role == CTRL_STANDBY && ha_heartbeat_is_fresh(ha)) + return 0; + + /* In catch-up → deny */ + if (atomic_load_explicit(&ha->in_catch_up, memory_order_acquire)) + return 0; + + /* Freshness gate requires knowing the candidate's config_tag/lsn. + * The raft library does not pass these; they are carried in the capnp + * RequestVoteRequest fields. For the adapter, freshness is checked in + * handle_request_vote (called from the iomgr decode path in task 6.3). + * Here we return 1 (eligible) so the raft library proceeds; the full + * freshness gate is enforced by the caller before invoking raft_on_vote_request. + * This is safe because the caller (task 6.3 dispatcher) already applies + * the freshness check from handle_request_vote before delegating here. */ + return 1; +} + +/* raft_election_timeout_fn: delaymgr callback for raft election timeout. + * Fires raft_on_election_timeout after delay_ms if not cancelled. + * Owns the election_timeout_ctx_t — frees it on exit. */ +static int raft_election_timeout_fn(void *arg) +{ + election_timeout_ctx_t *t = (election_timeout_ctx_t *)arg; + ha_state_t *ha = t->ha; + + /* Clear the stored pointer before checking cancelled so on_cancel_timeout + * is a safe no-op if it races with this callback. */ + ha->raft_election_timeout_ctx = NULL; + + if (atomic_load(&t->cancelled)) { + free(t); + return 0; + } + raft_on_election_timeout(&ha->raft); + free(t); + return 0; +} + +/* + * raft_cb_schedule_timeout: schedule election timeout via delaymgr. + * + * Creates an election_timeout_ctx_t and calls delay_task_ms. + * The pointer is stored in ha->raft_election_timeout_ctx for cancellation. + */ +static void raft_cb_schedule_timeout(void *ctx, uint32_t delay_ms) +{ + ha_state_t *ha = (ha_state_t *)ctx; + cluster_t *c = ha->raft_c; + if (!c) return; /* test mode */ + + election_timeout_ctx_t *t = calloc(1, sizeof(*t)); + if (!t) return; + + atomic_init(&t->cancelled, 0); + t->ha = ha; + t->c = c; + t->ctx = NULL; /* raft path does not use ha_elect_ctx_t */ + + int rc = delay_task_ms(&c->delaymgr, (int)delay_ms, 1, + raft_election_timeout_fn, t); + if (rc < 0) { + WARN("raft_cb_schedule_timeout: delay_task_ms failed: %d", rc); + free(t); + return; + } + + ha->raft_election_timeout_ctx = t; +} + +/* + * raft_cb_cancel_timeout: cancel the pending election timeout. + * Sets cancelled=1 on the stored election_timeout_ctx_t. + */ +static void raft_cb_cancel_timeout(void *ctx) +{ + ha_state_t *ha = (ha_state_t *)ctx; + election_timeout_ctx_t *t = + (election_timeout_ctx_t *)ha->raft_election_timeout_ctx; + if (t) + atomic_store(&t->cancelled, 1); + ha->raft_election_timeout_ctx = NULL; +} + +/* raft_rdhup_delay_fn: delaymgr callback for RDHUP-triggered delayed election. + * Owns its ctx (election_timeout_ctx_t) — frees it on exit. + * Calls raft_check_and_start_election to start election only if not already voted. */ +static int raft_rdhup_delay_fn(void *arg) +{ + election_timeout_ctx_t *t = (election_timeout_ctx_t *)arg; + ha_state_t *ha = t->ha; + if (!atomic_load(&t->cancelled)) + raft_check_and_start_election(&ha->raft); + free(t); + return 0; +} + +/* + * raft_cb_start_election_after_delay: schedule RDHUP retry timer. + * After delay_ms, calls raft_check_and_start_election to start election + * only if this node hasn't already voted for another candidate. + */ +static void raft_cb_start_election_after_delay(void *ctx, uint32_t delay_ms) +{ + ha_state_t *ha = (ha_state_t *)ctx; + cluster_t *c = ha->raft_c; + if (!c) return; /* test mode */ + + election_timeout_ctx_t *t = calloc(1, sizeof(*t)); + if (!t) return; + + atomic_init(&t->cancelled, 0); + t->ha = ha; + t->c = c; + t->ctx = NULL; + + int rc = delay_task_ms(&c->delaymgr, (int)delay_ms, 1, + raft_rdhup_delay_fn, t); + if (rc < 0) { + WARN("raft_cb_start_election_after_delay: delay_task_ms failed: %d", rc); + free(t); + } +} + +/* + * raft_cb_become_leader: node won the election — enter COMMITTING state. + * + * Equivalent of ha_elect_evaluate_and_signal CTRL_PRIMARY path: + * - enter CTRL_PRIMARY_COMMITTING + * - call cluster_init_phase2 on first boot + * - reset repl_ack_count, push config, send ctrlhello + * - arm committing timeout watchdog + * + * ha->raft_c must be non-NULL (production path only; test mode uses noop stubs). + */ +static void raft_cb_become_leader(void *ctx, uint64_t term) +{ + ha_state_t *ha = (ha_state_t *)ctx; + cluster_t *c = ha->raft_c; + + /* Cancel any pending election timeout */ + raft_cb_cancel_timeout(ctx); + + /* Sync ha->term from raft term (raft owns term increments) */ + atomic_store_explicit(&ha->term, term, memory_order_release); + + int is_first_boot = !ha->phase2_done; + + /* First boot: call cluster_init_phase2 */ + if (is_first_boot) { +#ifdef XCE_TESTING + 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 */ + atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY_COMMITTING, + memory_order_release); + INFO("HA FSM (raft): → CTRL_PRIMARY_COMMITTING (term=%" PRIu64 + ", first_boot=%d)", term, is_first_boot); + + /* Reset ack counter for this committing round */ + atomic_store(&ha->repl_ack_count, 0); + + if (c) { + /* Push config to standbys with lower config_tag */ +#ifdef XCE_TESTING + if (ha_push_config_hook != NULL) + ha_push_config_hook(c); +#else + push_config_to_standbys(c, NULL, 0, NULL, 0); +#endif + /* Send CtrlHelloRequest to all standbys to trigger WAL catch-up */ +#ifdef XCE_TESTING + if (ha_send_ctrlhello_hook != NULL) + ha_send_ctrlhello_hook(c); +#else + ha_send_ctrlhello_to_all_standbys(c); +#endif + /* Arm committing timeout watchdog */ + 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("raft_cb_become_leader: delay_task_ms for committing timeout failed: %d", + rc); + free(ct); + } + } + } +} + +/* + * raft_cb_become_follower: node lost election or saw higher term — enter STANDBY. + * + * Syncs ha->term from the raft-reported term, sets role to CTRL_STANDBY, + * calls cluster_init_phase2 on first boot (standbys also need it to + * receive config replication push from the new primary). + */ +static void raft_cb_become_follower(void *ctx, uint64_t term, int leader_idx) +{ + (void)leader_idx; + ha_state_t *ha = (ha_state_t *)ctx; + cluster_t *c = ha->raft_c; + + /* Cancel any pending election timeout */ + raft_cb_cancel_timeout(ctx); + + /* Sync ha->term */ + atomic_store_explicit(&ha->term, term, memory_order_release); + + int is_first_boot = !ha->phase2_done; + + /* First boot: call cluster_init_phase2 so structures are ready */ + if (is_first_boot) { +#ifdef XCE_TESTING + 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 (raft): → CTRL_STANDBY (term=%" PRIu64 ", first_boot=%d)", + term, is_first_boot); + + /* Trigger async config fetch from primary so standby can load config */ +#ifndef XCE_TESTING + if (c != NULL && is_first_boot) { + extern int fetch_config_from_primary(cluster_t *c); + delay_task_ms(&c->delaymgr, 100, 1, + (delay_task_entry_fn_t)fetch_config_from_primary, c); + } +#endif +} + +/* + * raft_cb_persist: persist term + voted_for to disk before granting vote. + * + * Maps raft integer voted_for index to a candidate name using cluster config. + * Updates ha->term, ha->voted_for_term, ha->voted_for_name then calls + * ha_persist_term(ha, workdir). + * + * voted_for == -1 means "no vote"; voted_for >= 0 is the candidate index. + * Returns 0 on success, non-zero on error (causes raft library to deny vote). + */ +static int raft_cb_persist(void *ctx, uint64_t term, int voted_for) +{ + ha_state_t *ha = (ha_state_t *)ctx; + cluster_t *c = ha->raft_c; + const char *workdir = (c && c->myconf) ? c->myconf->workdir : NULL; + + /* Update ha->term */ + atomic_store_explicit(&ha->term, term, memory_order_release); + + /* Map voted_for index to name */ + pthread_mutex_lock(&ha->lock); + + if (voted_for == -1) { + ha->voted_for_term = 0; + memset(ha->voted_for_name, 0, sizeof(ha->voted_for_name)); + } else { + ha->voted_for_term = term; + /* Look up candidate name by index */ + if (c && c->myconf && + voted_for < c->myconf->peers_cfg.candidates_count) { + const char *name = c->myconf->peers_cfg.candidates[voted_for].name; + strncpy(ha->voted_for_name, name, sizeof(ha->voted_for_name) - 1); + ha->voted_for_name[sizeof(ha->voted_for_name) - 1] = '\0'; + } else { + /* Test mode or out-of-range: use "self" placeholder */ + if (voted_for == ha->my_priority) { + strncpy(ha->voted_for_name, "self", + sizeof(ha->voted_for_name) - 1); + } else { + memset(ha->voted_for_name, 0, sizeof(ha->voted_for_name)); + } + } + } + + pthread_mutex_unlock(&ha->lock); + + ha_persist_term(ha, workdir); + return 0; +} + +static const raft_callbacks_t s_real_cbs = { + .on_send_vote_request = raft_cb_send_vote_request, + .on_vote_eligible = raft_cb_vote_eligible, + .on_become_leader = raft_cb_become_leader, + .on_become_follower = raft_cb_become_follower, + .on_persist = raft_cb_persist, + .on_schedule_timeout = raft_cb_schedule_timeout, + .on_cancel_timeout = raft_cb_cancel_timeout, + .on_start_election_after_delay = raft_cb_start_election_after_delay, .on_trace = NULL, }; @@ -127,30 +588,33 @@ int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority) * ha_wait_for_heartbeat_or_timeout() calls sem_timedwait on this. */ sem_init(&ha->heartbeat_seen, 0, 0); - /* ── Task 6.1: Initialize embedded raft_state_t ────────────────────────── + /* ── Task 6.1/6.2: Initialize embedded raft_state_t ───────────────────── * - * Wire raft_init with stub (no-op) callbacks — real callbacks are connected - * in task 6.2. initial_term and initial_voted_for start at defaults (0 / -1) - * here; ha_restore_term() updates ha->term from disk immediately after ha_init - * in cluster.c, so the raft state will be reconciled before any election fires. + * Wire real callbacks (s_real_cbs) so the state machine is ready when + * ha_start_election_async re-initialises raft with the correct n_peers / + * self_idx / initial_term after ha_restore_term(). * - * cfg.ctx = ha so task 6.2 callbacks can reach ha_state_t. - * cfg.n_peers = 1 (single-node default) and cfg.self_idx = 0 are safe placeholders; - * ha_start_election_async already stores the real peer count in ha->n_peers - * before calling any Raft function. + * n_peers = 1 and self_idx = 0 are safe placeholders here; they are + * overwritten by ha_raft_reinit() (called from ha_start_election_async) + * which runs after ha_restore_term() and knows the real cluster size. * - * raft_init requires non-NULL required callbacks; task 6.2 replaces these. - * The no-op stubs are defined at file scope above (s_noop_cbs). + * initial_term = 0 / initial_voted_for = -1 are first-boot defaults; + * ha_raft_reinit() restores the persisted values before any election fires. + * + * ha->raft_c is NULL until ha_start_election_async sets it. */ + ha->raft_c = NULL; + ha->raft_election_timeout_ctx = NULL; + raft_config_t raft_cfg; memset(&raft_cfg, 0, sizeof(raft_cfg)); - raft_cfg.n_peers = 1; /* placeholder; real count set by ha_start_election_async */ - raft_cfg.self_idx = 0; /* placeholder; real index set in task 6.2 */ + raft_cfg.n_peers = 1; /* placeholder; overwritten by ha_raft_reinit */ + raft_cfg.self_idx = my_priority; /* self_idx == priority by convention */ raft_cfg.election_timeout_ms = cfg->election_timeout_ms; raft_cfg.my_priority = my_priority; - raft_cfg.initial_term = 0; /* ha_restore_term() updates ha->term after ha_init */ - raft_cfg.initial_voted_for = -1; /* no vote on fresh init */ - raft_cfg.cbs = &s_noop_cbs; + raft_cfg.initial_term = 0; /* ha_restore_term() updates ha->term after ha_init */ + raft_cfg.initial_voted_for = -1; /* no vote on fresh init */ + raft_cfg.cbs = &s_real_cbs; raft_cfg.ctx = ha; raft_init(&ha->raft, &raft_cfg); @@ -990,46 +1454,15 @@ 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. +/* ── Test-only hook shims (XCE_TESTING only) ────────────────────────────── + * Type definitions (election_timeout_ctx_t, committing_timeout_ctx_t), + * extern declarations (cluster_init_phase2, push_config_to_standbys, + * ha_send_ctrlhello_to_all_standbys), and static forward declarations + * (committing_timeout_fn, ha_heartbeat_is_fresh) have been moved to the + * forward-declarations block near the top of this file so that the raft + * callback implementations (Task 6.2) can reference them. */ -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); -/* ha_send_ctrlhello_to_all_standbys: defined in cluster.c. - * Iterates ha->standby_endpoints, sends CtrlHelloRequest to each, triggering - * catch-up replication in handle_ctrlhello (Req 7.2). */ -extern void ha_send_ctrlhello_to_all_standbys(cluster_t *c); -/* push_config_to_standbys: defined in config_repl.c (Req 7.1). - * Best-effort fan-out of config blob to all connected standbys. - * Called with (c, NULL, 0, NULL, 0) to push from the primary's saved config. */ -extern int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, - const uint8_t sha256[32], uint64_t new_tag); -#endif - #ifdef XCE_TESTING /* * ha_phase2_hook: test-only seam. @@ -1350,16 +1783,7 @@ void handle_request_vote_reply(ha_state_t *ha, * endpoint_t.ssl_handshake_done transitions to 1 in io_handler). */ -typedef struct { - ha_state_t *ha; - cluster_t *c; - ha_elect_ctx_t *ctx; - int peer_idx; - uint64_t my_term; - uint64_t my_tag; - uint64_t my_lsn; - const char *my_name; -} vote_ep_ctx_t; +/* vote_ep_ctx_t: typedef moved to Task 6.2 forward-declaration block above */ /* xce_request: defined in request.c, not yet in any public header */ extern int xce_request(endpoint_t *ep, xce_request_t *req); @@ -1851,6 +2275,49 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c, uint64_t my_config_tag, uint64_t my_lsn, const char *my_name) { + /* Step 0: wire cluster pointer for raft callbacks (task 6.2). + * 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); ha_start_election_async + * receives n_peers as the count EXCLUDING self, so total = n_peers+0 + * is wrong — n_peers already IS the total candidates_count when + * called from cluster.c. Use it directly for raft_state_t.n_peers. + * + * 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 1: store peer count */ ha->n_peers = n_peers; diff --git a/src/controller/ha.h b/src/controller/ha.h index 0f7f55f..16bcd38 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -176,6 +176,9 @@ typedef struct { int n_peers; /* total candidate count (excluding self) */ /* ── New fields added by raft-library spec (Task 6.1) ──────── */ raft_state_t raft; /* Raft consensus state (wired in task 6.2) */ + /* ── New fields added by raft-library spec (Task 6.2) ──────── */ + cluster_t *raft_c; /* cluster_t* set by ha_start_election_async; used by raft callbacks */ + void *raft_election_timeout_ctx; /* election_timeout_ctx_t* for cancel (task 6.2) */ } ha_state_t; /* ── Public API ────────────────────────────────────────────── */