myxce/src/controller/ha.h

457 lines
20 KiB
C

#ifndef HA_H
#define HA_H
/*
* XCE Controller HA — role management and election (P1).
*
* P1-1 audit: the existing create_ssl_context() + iomgr_set_ssl_ctx() path
* in cluster.c (line 559, set at line 583) covers all controller-to-controller SSL.
* HA listeners (heartbeat, election probe, replication accept) reuse the same
* ssl_ctx already installed on c->iomgr.
* No new TLS infrastructure is needed — XCE runs on a trusted management LAN.
* Peer identity is validated by host:port match against the --peers candidate list.
* See design doc Part 5.9 and memory: Persistence & HA Trusted LAN Decision.
*
* P1-2: full struct definitions and API.
* endpoint_t* (primary_ep) is intentionally omitted here — including netio.h
* would pull in heavy OpenSSL headers and create circular dependency risks.
* The watchdog (P1-3) will include netio.h directly where it needs endpoint_t.
*/
#include <pthread.h>
#include <semaphore.h>
#include <stdbool.h>
#include <stdatomic.h>
#include <stdint.h>
#include <time.h>
#include "config.h" /* XCE_MAXNAME_LEN, MAX_STANDBYS, etc. */
#include "xce/cluster.h" /* xce_candidate_t */
#include "zarray.h"
/* Forward declaration — avoids circular include with cluster_def.h */
typedef struct cluster_s cluster_t;
/* ── Inline PRNG for election jitter ───────────────────────── */
static inline uint64_t xorshift64_next(uint64_t *state)
{
uint64_t x = *state;
x ^= x << 13;
x ^= x >> 7;
x ^= x << 17;
*state = x;
return x;
}
/*
* ha_seed_jitter: seed the xorshift64 PRNG from /dev/urandom (8 bytes).
* Falls back to CLOCK_MONOTONIC ^ priority ^ pid if urandom is unavailable.
* Not inline — contains syscalls (open/read/close/clock_gettime/getpid).
*/
void ha_seed_jitter(uint64_t *seed, int my_priority);
/* ── Role enum ─────────────────────────────────────────────── */
typedef enum {
CTRL_PRIMARY = 0,
CTRL_STANDBY = 1,
CTRL_INIT = 2, /* Startup: listener open, awaiting heartbeat or election timeout */
CTRL_ELECTING = 3 /* Running quorum election via OP_requestVote */
} ctrl_role_t;
/* ── Election types ────────────────────────────────────────── */
/*
* ha_vote_entry_t: per-peer state accumulated by the election candidate.
* One entry per peer in ha_elect_ctx_t.votes[] (flexible array).
*/
typedef struct {
xce_candidate_t *peer; /* Peer being solicited */
uint64_t my_term; /* Term sent in OP_requestVote */
uint64_t my_tag; /* config_tag sent in request */
uint64_t my_lsn; /* WAL LSN sent in request */
int vote_granted; /* 1 if peer granted vote */
int replied; /* 1 if peer replied (granted or denied) */
uint64_t peer_tag; /* peer's config_tag from reply */
uint64_t peer_lsn; /* peer's WAL LSN from reply */
} ha_vote_entry_t;
/*
* ha_elect_ctx_t: shared context for one quorum election round.
* Allocated as: calloc(1, sizeof(*ctx) + (n_peers-1) * sizeof(ctx->votes[0]))
* Stored in ha_state_t.elect_ctx (void * _Atomic) during election; NULL otherwise.
*/
typedef struct {
int n_peers; /* Total peer count (excluding self) */
int votes_granted; /* Votes received; starts at 1 (self-vote) */
int n_replied; /* Number of peers that replied */
uint64_t max_resp_tag; /* Highest config_tag seen in any reply */
uint64_t max_resp_lsn; /* Highest WAL LSN seen among max_resp_tag peers */
int result; /* 1=primary, 0=standby, <0=error */
int done_posted; /* Guard: non-zero once sem_post(&done) has been called */
pthread_mutex_t lock; /* Protects votes_granted/n_replied/max_resp_tag/max_resp_lsn/result */
sem_t done; /* Startup thread blocks; dispatch posts when done */
ha_vote_entry_t votes[]; /* C99 flexible array: n_peers-1 entries */
} ha_elect_ctx_t;
/* ── HA configuration ──────────────────────────────────────── */
/* Default values used by cluster_init(). Tuned for a LAN with ≤5ms RTT. */
#define HA_DEFAULT_HEARTBEAT_INTERVAL_MS 3000
#define HA_DEFAULT_HEARTBEAT_TIMEOUT_MS 10000
#define HA_DEFAULT_ELECTION_TIMEOUT_MS 2000
#define HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS 500
#define HA_DEFAULT_MAX_PROMOTION_RETRIES 5
#define HA_DEFAULT_ELECTION_JITTER_MAX_MS 200
#define HA_DEFAULT_ELECTION_PRIORITY_STEP_MS 500
typedef struct {
uint32_t heartbeat_interval_ms; /* Primary heartbeat send interval (min 2000 ms) */
uint32_t heartbeat_timeout_ms; /* Standby timeout before promoting (>= interval) */
uint32_t election_timeout_ms; /* Election probe shared deadline (min 500 ms) */
uint32_t probe_connect_timeout_ms; /* Per-probe TCP connect timeout */
uint32_t max_promotion_retries; /* Max consecutive failed promotion attempts */
uint32_t election_jitter_max_ms; /* random jitter added to election timeout (default 200) */
uint32_t election_priority_step_ms; /* extra ms per priority level (default 500) */
} ha_config_t;
typedef struct {
xce_candidate_t *candidate; /* peer to probe */
void *ssl_ctx; /* SSL_CTX* (void* to avoid openssl header) */
uint32_t timeout_ms;
atomic_int *election_done;
int got_ack;
pthread_t thread;
} probe_ctx_t;
/* ── HA runtime state (embedded in cluster_t as `ha_state_t ha` in P1-5) ── */
typedef struct {
_Atomic uint64_t term; /* Election term — fencing token */
_Atomic int32_t role; /* Cast to ctrl_role_t at access sites */
int my_priority; /* Index in --peers list (lower = higher priority) */
_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 */
_Atomic int promotion_failed;
_Atomic int promotion_retry_count;
_Atomic int in_catch_up; /* Non-zero while processing historical WAL entries */
_Atomic int failover_requested; /* Set by SIGUSR1 handler; watchdog executes */
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 */
/* Watchdog thread — started in P1-3 */
_Atomic int watchdog_stop;
pthread_t watchdog_thread;
void *ssl_ctx; /* SSL_CTX* cast; avoids pulling openssl headers here */
uint64_t jitter_seed; /* xorshift64 PRNG state for election jitter */
zarray_t *standby_endpoints; /* connected standby controller endpoints for config push */
/* ── New fields added by ha-config-bootstrap spec ──────── */
uint64_t voted_for_term; /* Req 9: term of last granted vote */
char voted_for_name[XCE_MAXNAME_LEN]; /* Req 9: candidate of last granted vote */
int primary_ctrl_fd; /* Req 6: persistent channel fd (-1 = none) */
bool defer_config_fetch; /* Req 3: tag=0 node defers fetch to post-election */
void * _Atomic elect_ctx; /* Req 2: ha_elect_ctx_t* during election; NULL otherwise */
_Atomic uint64_t last_known_committed_lsn; /* Req 5: from primary heartbeats */
sem_t heartbeat_seen; /* Req 1: posted by heartbeat handler during CTRL_INIT */
uint64_t last_known_committed_config_tag; /* Req 8: from primary heartbeats */
_Atomic uint64_t committed_config_tag; /* Req 8.1: set on quorum ACK */
} ha_state_t;
/* ── Public API ────────────────────────────────────────────── */
/*
* ha_init: zero-initialise *ha, validate cfg, store config and my_priority.
* Returns 0 on success, -EINVAL on bad config.
* Initial role is always CTRL_STANDBY.
*/
int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority);
/*
* ha_fini: destroy mutex. Watchdog thread join added in P1-3.
*/
void ha_fini(ha_state_t *ha);
/*
* am_i_primary: stub — priority-0 node is primary.
* Full election probing added in P1-3.
*/
int am_i_primary(cluster_t *c);
/*
* ha_promote_to_primary: stub — atomically sets role to PRIMARY.
* Full promotion sequence (hooks, WAL flush, peer notify) added in P1-3a.
*/
int ha_promote_to_primary(cluster_t *c);
/*
* ha_demote_to_standby: stub — atomically sets role to STANDBY.
* Full demotion sequence added in P1-3a.
*/
int ha_demote_to_standby(cluster_t *c);
/* ── Internal promotion/demotion (P1-3a) ───────────────────── */
/*
* ha_do_promote: increment term, set role=PRIMARY, reset last_received_crc.
* No cluster_t dependency — fully testable standalone.
* Callers: ha_promote_to_primary() (P1-5), watchdog failover path.
*/
void ha_do_promote(ha_state_t *ha);
/*
* ha_do_demote: set role=STANDBY, reset hooks_registered idempotency guard.
* Hook disable requires cluster_t and is done in ha_demote_to_standby() (P1-5).
*/
void ha_do_demote(ha_state_t *ha);
/*
* ha_install_sigusr1: register a SIGUSR1 handler that atomically sets
* ha->failover_requested. The handler uses only async-signal-safe operations
* (atomic store on an int-width type — POSIX.1-2017 §2.4.3).
* Must be called once, before any signal can arrive.
*/
void ha_install_sigusr1(ha_state_t *ha);
/* ── Watchdog thread API (P1-3) ────────────────────────────── */
/*
* ha_start_watchdog: spawn the heartbeat watchdog pthread.
* Must be called after ha_init(). Returns 0 on success, -errno on failure.
*/
int ha_start_watchdog(ha_state_t *ha);
/*
* ha_stop_watchdog: set watchdog_stop flag and pthread_join the thread.
* Safe to call if watchdog was never started (thread == 0).
*/
void ha_stop_watchdog(ha_state_t *ha);
/*
* ha_check_heartbeat_nocluster: test-only entry point — logic is inlined in
* ha_watchdog_thread() in production. Only compiled under XCE_TESTING.
*/
#ifdef XCE_TESTING
void ha_check_heartbeat_nocluster(ha_state_t *ha);
#endif
/* ── Redirect helpers used by join.c ──────────────────────── */
/*
* ha_should_redirect: returns 1 if this controller should redirect join
* requests to the primary.
* Returns 0 if: (a) role is PRIMARY, or (b) primary_host is empty /
* primary_port is 0 (no primary known yet — client should retry).
*/
int ha_should_redirect(const ha_state_t *ha);
/*
* ha_no_primary_known: returns 1 if role is STANDBY and primary_host is
* empty or primary_port is 0. Caller should reply E_XCE_NOMASTER so the
* client knows to retry later rather than treating the error as permanent.
*/
int ha_no_primary_known(const ha_state_t *ha);
/*
* ha_is_primary: returns 1 if ha->role is CTRL_PRIMARY, 0 otherwise.
* Lock-free. Safe to call from any thread.
*/
int ha_is_primary(const ha_state_t *ha);
/* ha_check_priority_stepdown() deleted — replaced by quorum election (Req 2.1) */
/*
* start_heartbeat_listener: start a minimal pre-Phase-2 SSL listener
* that responds to OP_heartbeat probes. Called after am_i_primary()
* returns 1. Records tid in c->heartbeat_listener_tid.
*/
void start_heartbeat_listener(cluster_t *c);
/*
* stop_heartbeat_listener: signal and join the heartbeat listener thread.
* Must be called before iomgr_init() to free the listening port (EADDRINUSE).
*/
void stop_heartbeat_listener(cluster_t *c);
/*
* ha_dispatch_gate: role-based operation filter for the vtaskmgr dispatch loop.
* Returns:
* 0 — operation is permitted for the current role
* -E_XCE_NOT_READY — role is CTRL_INIT or CTRL_ELECTING and opcode is not
* OP_requestVote or OP_probeConfigTag
* -E_XCE_STANDBY — role is CTRL_STANDBY and opcode is a primary-only op
* CTRL_PRIMARY always returns 0.
*/
int ha_dispatch_gate(ha_state_t *ha, int opcode);
/* ── HA Config Bootstrap & Term Durability (ha-config-bootstrap spec) ── */
/*
* ha_persist_term: write ha-term.bin atomically to {workdir}/ha-term.bin.
* Uses write + rename for durability (existing pattern in the codebase).
* Must be called before any vote is granted or term is incremented.
* If workdir is NULL, the file write is skipped (safe for unit tests).
*/
void ha_persist_term(ha_state_t *ha, const char *workdir);
/*
* ha_restore_term: read {workdir}/ha-term.bin at startup and restore
* ha->term and ha->voted_for_* into ha_state_t.
*
* On missing file (ENOENT): no-op — term stays 0, voted_for_name stays empty.
* Returns 0 (safe first-boot path). Req 9.3.
* On CRC/magic mismatch: log WARN, zero ha->term and ha->voted_for_* for
* safety and return 0. Req 9.2.
* On valid file: atomically restore ha->term; copy voted_for_term
* and voted_for_name. Returns 0. Req 9.2.
* On unexpected I/O error: return -errno.
*
* Must be called after ha_init() and before any election or vote processing.
* If workdir is NULL, the function is a no-op (safe for unit tests).
* Requirements: 9.1, 9.2, 9.3.
*/
int ha_restore_term(ha_state_t *ha, const char *workdir);
/*
* handle_request_vote: process an incoming OP_requestVote RPC.
* Applies term/voted_for/freshness gate; persists state via ha_persist_term().
* Returns 1 = vote granted, 0 = vote denied, <0 = error.
* Requirements: 2.2, 2.3, 2.4, 2.5, 2.6, 9.1
*/
int handle_request_vote(ha_state_t *ha,
uint64_t term,
const char *candidate_id,
uint64_t config_tag,
uint64_t last_lsn,
uint64_t my_config_tag,
uint64_t my_last_lsn,
int in_catch_up_flag);
/*
* ha_elect_evaluate_and_signal: evaluate accumulated ha_elect_ctx_t votes.
* Sets ctx->result (1=primary, 0=standby) and posts ctx->done semaphore.
* Applies quorum check, fresher-peer yield gate, and config_tag=0 fatal gate.
* Requirements: 2.1, 2.7, 2.8
*/
ctrl_role_t ha_elect_evaluate_and_signal(ha_state_t *ha, ha_elect_ctx_t *ctx);
/*
* am_i_primary_quorum: top-level Raft-style quorum election entrypoint.
* Single-node: self-elects immediately (Req 2.9). Multi-node: spawns per-peer
* vote-request threads, blocks on ctx->done, retries once on split vote (Req 2.10).
* Returns CTRL_PRIMARY or CTRL_STANDBY.
* Requirements: 2.1, 2.9, 2.10, 2.11
*/
ctrl_role_t am_i_primary_quorum(ha_state_t *ha,
int n_peers,
xce_candidate_t **peers,
uint64_t my_config_tag,
uint64_t my_lsn);
/*
* handle_request_vote_reply: process an incoming reply to an OP_requestVote RPC.
*
* Called by each per-peer vote-request thread once the peer replies (or times out).
* Updates the entry in ctx->votes[peer_idx] with the reply fields.
* Increments ctx->votes_granted if vote_granted is non-zero.
* Increments ctx->n_replied unconditionally.
* Updates ctx->max_resp_tag / ctx->max_resp_lsn to track the freshest peer seen.
* When quorum is reached (votes_granted >= quorum) OR all peers have replied
* (n_replied >= n_peers), calls ha_elect_evaluate_and_signal() to post ctx->done.
*
* peer_idx: 0-based index into ctx->votes[] identifying the replying peer.
* term: term the peer reported in its reply (used for fencing if > ha->term).
* vote_granted: non-zero if the peer granted the vote, 0 if denied.
* peer_config_tag: peer's current config_tag at reply time.
* peer_lsn: peer's current WAL LSN at reply time.
*
* Requirements: 2.2, 2.3, 2.4, 2.5, 2.6, 9.1
*/
void handle_request_vote_reply(ha_state_t *ha,
ha_elect_ctx_t *ctx,
int peer_idx,
uint64_t term,
int vote_granted,
uint64_t peer_config_tag,
uint64_t peer_lsn);
/*
* bootstrap_config_if_needed: pre-election config probe.
* If the local config tag is 0 and at least one peer has config, sets
* ha->defer_config_fetch = true and returns without fetching.
* The actual fetch is deferred to post-election (after a winner is known).
* Returns 0 on success, -errno on unrecoverable error.
*/
int bootstrap_config_if_needed(cluster_t *c);
/*
* xce_probe_config_tag: single-RTT OP_probeConfigTag RPC to one peer.
* Fills *tag_out with the peer's current config_tag on success.
* Returns 0 on success, -errno on connection/protocol error.
*/
int xce_probe_config_tag(const char *host, uint16_t port, uint64_t *tag_out);
/*
* ha_wait_for_heartbeat_or_timeout: wait in CTRL_INIT for an incoming heartbeat.
*
* Blocks on sem_timedwait(&ha->heartbeat_seen, &deadline) where
* deadline = now(CLOCK_REALTIME) + election_timeout_ms + xorshift64_jitter
*
* Returns CTRL_STANDBY — heartbeat arrived; ha_do_demote() was called
* (role is now CTRL_STANDBY).
* Returns CTRL_ELECTING — deadline expired; caller must drive the election.
*
* The heartbeat handler posts heartbeat_seen only when role == CTRL_INIT.
* OPCode_heartbeat is allowed through ha_dispatch_gate() in CTRL_INIT state
* so that the primary's periodic heartbeat can reach the handler.
*
* Requirements: 1.1 (heartbeat → CTRL_STANDBY), 1.2 (timeout → CTRL_ELECTING).
* Boundary: Controller FSM (ha.c).
*/
ctrl_role_t ha_wait_for_heartbeat_or_timeout(ha_state_t *ha);
/* ── Task 9.2: Persistent Control Channel (Req 6.1) ───────────────────────── */
/*
* ha_find_and_connect_to_primary: three-tier primary discovery and raw TCP connect.
*
* Tier 1: ha->voted_for_name non-empty → direct connect to that candidate.
* Backoff HA_FIND_PRIMARY_T1_BACKOFF_MS; retry HA_FIND_PRIMARY_T1_MAX_RETRIES.
* Tier 2: probe remaining candidates; follow XCE_ERR_STANDBY redirect.
* Tier 3: blind scan with 100ms→5s exponential backoff (no exit until success).
*
* On success: sets ha->primary_ctrl_fd; when c != NULL, registers fd with
* c->iomgr.epoll_fd for EPOLLIN|EPOLLRDHUP. Returns fd >= 0.
* Pass c == NULL to skip iomgr registration (unit test mode).
* Requirements: 6.1
*/
int ha_find_and_connect_to_primary(ha_state_t *ha,
const ha_config_t *cfg,
const xce_cluster_config_t *peers,
cluster_t *c);
#define HA_FIND_PRIMARY_T1_MAX_RETRIES 3
#define HA_FIND_PRIMARY_T1_BACKOFF_MS 200
#define HA_FIND_PRIMARY_T3_INIT_MS 100
#define HA_FIND_PRIMARY_T3_MAX_MS 5000
/* ── P3-6: Test-only failover shim ────────────────────────────── */
#ifdef XCE_TESTING
/* Test-only shim: demote then re-promote; returns new term.
* Exercises the manual failover state machine without a live cluster_t.
* Calls ha_do_demote() then ha_do_promote(); validates role transitions
* and term monotonicity. See design doc P3-6 and src/docs/persistence-and-ha-design.md.
*/
uint64_t ha_trigger_manual_failover_for_test(ha_state_t *ha);
/*
* ha_tcp_connect_hook: test-only seam for TCP connections.
* When non-NULL, ha_find_and_connect_to_primary() calls this instead of
* the real connect() path. Returns connected fd on success, -errno on failure.
*/
extern int (*ha_tcp_connect_hook)(const char *host, int port);
#endif
#endif /* HA_H */