feat(raft-library): task 6.2 — real raft callbacks in ha.c; fix double-free + dead stubs
This commit is contained in:
parent
b77525deb0
commit
8f0d2d55c3
3 changed files with 552 additions and 82 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
||||
|
|
|
|||
|
|
@ -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 ────────────────────────────────────────────── */
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue