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

feat(raft-library): task 6.2 — real raft callbacks in ha.c; fix double-free + dead stubs

Browse source
This commit is contained in:
Rongsong Shen 2026-05-31 07:53:00 +00:00
parent b77525deb0
commit 8f0d2d55c3
3 changed files with 552 additions and 82 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.kiro/specs/raft-library/tasks.md
View file

@ -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

629
src/controller/ha.c
View file

@ -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;

3
src/controller/ha.h
View file

@ -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 ────────────────────────────────────────────── */