Rongsong Shen
18ad3e9c9f
- Add ha_repl_send_snapshot_hook seam in ha.c/ha.h for XCE_TESTING - test_dispatch_repl_action_send_snapshot: verifies SEND_SNAPSHOT decision triggers hook when standby_lsn < oldest_wal_lsn - test_snapshot_recv_two_chunks_assembles_and_renames: verifies standby assembles snap.part across chunks and renames on isLast=true - test_snapshot_recv_reconnect_discards_snappart: verifies O_TRUNC semantics on chunk_index==0 restart after reconnect - All tests pass under ASAN
722 lines
35 KiB
C
722 lines
35 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 "myconf.h" /* myconf_t, config_version_tag */
|
||
#include "persist.h" /* persist_t, apply_wal_entry */
|
||
#include "typemap.h" /* wal_catchup_entry_t, state_report_t, xce_reply_t */
|
||
#include "xce/cluster.h" /* xce_candidate_t */
|
||
#include "zarray.h"
|
||
#include "raft/raft.h" /* raft_state_t, raft_init */
|
||
|
||
/* 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_PRIMARY_COMMITTING = 4 /* Primary waiting for quorum ACK on config commit */
|
||
} 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).
|
||
*/
|
||
/* Forward declaration to avoid pulling netio.h into ha.h */
|
||
struct endpoint_t;
|
||
|
||
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 */
|
||
struct endpoint_t *ep; /* Outgoing vote endpoint; NULL if connect failed or not yet connected */
|
||
} 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 */
|
||
/* Fix 4 (HIGH): pointer back to the election_timeout_ctx_t so that task 4.1
|
||
* (quorum-reached path) can set timeout_ctx->cancelled = 1 to cancel the
|
||
* watchdog. Set by ha_start_election_async after scheduling the timeout. */
|
||
void *timeout_ctx; /* election_timeout_ctx_t*; NULL until watchdog scheduled */
|
||
/* Task 4.1: cluster_t* so ha_elect_evaluate_and_signal can call
|
||
* cluster_init_phase2(c) and arm the committing timeout watchdog.
|
||
* NULL in unit tests (c=NULL bypasses phase2 and watchdog arming). */
|
||
cluster_t *c;
|
||
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 5000
|
||
#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
|
||
#define HA_DEFAULT_COMMITTING_TIMEOUT_MS 10000
|
||
|
||
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) */
|
||
uint32_t committing_timeout_ms; /* Quorum-ACK deadline for CTRL_PRIMARY_COMMITTING (default 10000) */
|
||
} ha_config_t;
|
||
|
||
/* probe_ctx_t deleted — only used by am_i_primary() (election-iomgr-refactor) */
|
||
|
||
/* ── 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 */
|
||
/* failover_requested deleted — watchdog replaced by ha_arm_heartbeat_timeout_vtask */
|
||
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_stop / watchdog_thread deleted — replaced by ha_arm_heartbeat_timeout_vtask */
|
||
void *ssl_ctx; /* SSL_CTX* cast; avoids pulling openssl headers here */
|
||
uint64_t jitter_seed; /* xorshift64 PRNG state for election jitter */
|
||
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 */
|
||
/* ── New fields added by election-iomgr-refactor spec (Task 2.1) ── */
|
||
void *client_ssl_ctx; /* SSL_CTX* (TLS_client_method); created in myapp_run */
|
||
int phase2_done; /* 1 after cluster_init_phase2 runs */
|
||
_Atomic int repl_ack_count; /* count of standbys that ACKed in COMMITTING state */
|
||
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) */
|
||
void *committing_timeout_ctx; /* committing_timeout_ctx_t* for cancel on step-down */
|
||
/* ── New fields added by ha-heartbeat-liveness spec (Task 1.1) ── */
|
||
time_t last_activity; /* CLOCK_MONOTONIC seconds of last primary message; 0 = none yet */
|
||
/* ── New fields added by ha-repl-hardening spec (Task 1.4) ──────── */
|
||
uint64_t standby_reported_tag[MAX_HA_CANDIDATES]; /* Last config_tag reported by each standby */
|
||
uint64_t standby_reported_lsn[MAX_HA_CANDIDATES]; /* Last applied_lsn reported by each standby */
|
||
bool standby_converged[MAX_HA_CANDIDATES]; /* Whether standby[i] has converged with primary */
|
||
} 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() deleted — replaced by ha_start_election_async (election-iomgr-refactor) */
|
||
|
||
/*
|
||
* 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);
|
||
|
||
|
||
/* ── 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() / stop_heartbeat_listener() deleted — never called (election-iomgr-refactor) */
|
||
|
||
/*
|
||
* 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_apply_wal_catchup_entry: apply a single WAL catch-up entry to local
|
||
* persistent storage and advance ha->applied_lsn on success.
|
||
*
|
||
* Called from cluster.c OPCode_walCatchupEntry dispatch handler (Task 2.1).
|
||
* On success (rc == 0): applied_lsn advances to entry->lsn+1.
|
||
* On error (rc < 0): logs ERROR, returns error; caller skips (Req 3.3).
|
||
* applied_lsn is in-memory only — not persisted to disk.
|
||
* Requirements: 3.1, 3.2, 3.3, 3.4, 3.5
|
||
*/
|
||
int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
|
||
wal_catchup_entry_t *entry);
|
||
|
||
/* ha_nop_reply_state_report — populate a heartbeat reply with stateReport.
|
||
* Fills *sr from ha->applied_lsn (atomic read) and myconf->config_version_tag,
|
||
* then sets reply_out->data_which = stateReportMsg and
|
||
* reply_out->data.state_report = sr.
|
||
* Caller must keep *sr live until xce_reply() serialises the reply. */
|
||
void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf,
|
||
state_report_t *sr, xce_reply_t *reply_out);
|
||
|
||
/* ── Task 3.2: Unified replication decision enum and evaluator ────────────── */
|
||
|
||
/*
|
||
* ha_repl_action_t: result of ha_evaluate_repl_action().
|
||
*
|
||
* Encodes the single replication action the primary should take for a standby
|
||
* on the current heartbeat cycle, based solely on the standby's reported
|
||
* (config_tag, applied_lsn) vs the primary's committed state.
|
||
*
|
||
* Decision tree (design §Unified Heartbeat-Driven Protocol):
|
||
* standby_tag < committed_tag → PUSH_CONFIG
|
||
* standby_lsn < oldest_wal_lsn → SEND_SNAPSHOT
|
||
* standby_lsn < committed_lsn → STREAM_WAL
|
||
* tag >= committed AND lsn >= committed → CONVERGED
|
||
*
|
||
* Requirements: 2.2, 2.3, 2.4 (ha-repl-hardening spec)
|
||
*/
|
||
typedef enum {
|
||
HA_REPL_PUSH_CONFIG, /* standby config tag is behind — push config first */
|
||
HA_REPL_SEND_SNAPSHOT, /* standby LSN predates oldest WAL — transfer snapshot */
|
||
HA_REPL_STREAM_WAL, /* standby LSN is within WAL window — stream entries */
|
||
HA_REPL_CONVERGED, /* standby tag and LSN are both current */
|
||
} ha_repl_action_t;
|
||
|
||
/*
|
||
* ha_evaluate_repl_action: pure decision function — no side effects.
|
||
*
|
||
* Evaluates the standby's reported (standby_tag, standby_lsn) against the
|
||
* primary's committed state (committed_tag, committed_lsn, oldest_wal_lsn)
|
||
* and returns the single action the primary should take this heartbeat cycle.
|
||
*
|
||
* Arguments:
|
||
* standby_tag — config_tag last reported by this standby
|
||
* standby_lsn — applied_lsn last reported by this standby
|
||
* committed_tag — primary's committed config_tag
|
||
* committed_lsn — primary's committed WAL LSN
|
||
* oldest_wal_lsn — oldest retained WAL LSN (entries before this are gone)
|
||
*
|
||
* Requirements: 2.2, 2.3, 2.4 (ha-repl-hardening spec)
|
||
* Boundary: Primary Heartbeat Handler (ha.c)
|
||
*/
|
||
ha_repl_action_t ha_evaluate_repl_action(uint64_t standby_tag,
|
||
uint64_t standby_lsn,
|
||
uint64_t committed_tag,
|
||
uint64_t committed_lsn,
|
||
uint64_t oldest_wal_lsn);
|
||
|
||
/*
|
||
* ha_process_standby_state: update per-standby convergence tracking.
|
||
*
|
||
* Called on each heartbeat reply from a standby while the primary is in
|
||
* CTRL_PRIMARY_COMMITTING or CTRL_PRIMARY. Stores the standby's reported
|
||
* (config_tag, applied_lsn) and evaluates whether the standby has converged
|
||
* with the primary's committed state.
|
||
*
|
||
* Convergence criteria (Req 1.3):
|
||
* reported_tag >= committed_tag AND reported_lsn >= committed_lsn
|
||
*
|
||
* Logs INFO on the first transition from not-converged → converged (Req 7.2).
|
||
* Regression (converged → not-converged) clears standby_converged[peer_idx]
|
||
* and re-evaluates (Req 2.5).
|
||
*
|
||
* peer_idx: 0-based index into ha->standby_reported_* arrays; must be in
|
||
* [0, MAX_HA_CANDIDATES).
|
||
*
|
||
* Requirements: 1.2, 2.2, 2.5 (ha-repl-hardening spec)
|
||
* Boundary: Primary Heartbeat Handler (ha.c)
|
||
*/
|
||
void ha_process_standby_state(ha_state_t *ha, int peer_idx,
|
||
uint64_t reported_tag, uint64_t reported_lsn,
|
||
uint64_t committed_tag, uint64_t committed_lsn);
|
||
|
||
/* ── Task 3.5c: Standby snapshot receive handler ──────────────────────────── */
|
||
|
||
/*
|
||
* ha_handle_snapshot_chunk: receive one snapshot transfer chunk on the standby.
|
||
*
|
||
* Each chunk's payload bytes are written (or appended) to snap.part in snap_dir.
|
||
* When chunk->is_last is non-zero:
|
||
* 1. Verify total bytes written == chunk->total_size.
|
||
* 2. Atomically rename snap.part → snap-{snapshot_lsn:016x}.bin.
|
||
* 3. Load the snapshot via persist_load_snapshot().
|
||
* 4. Set ha->applied_lsn = p->checkpoint_lsn (in-memory only).
|
||
*
|
||
* On mid-transfer reconnect (caller resets state), snap.part is discarded.
|
||
* Returns 0 on success, negative errno on error.
|
||
*
|
||
* Requirements: 4.2, 4.3, 4.4 (ha-repl-hardening spec)
|
||
* Boundary: Standby Dispatch (cluster.c, ha.c)
|
||
*/
|
||
int ha_handle_snapshot_chunk(ha_state_t *ha, persist_t *p,
|
||
const char *snap_dir,
|
||
snapshot_transfer_t *chunk,
|
||
const uint8_t *payload_bytes,
|
||
size_t payload_len);
|
||
|
||
/* ── 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);
|
||
|
||
/*
|
||
* 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);
|
||
|
||
/*
|
||
* 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.4: EPOLLRDHUP Failover and Heartbeat Timeout (Req 6.2–6.4, 6.8) ─
|
||
|
||
* ha_on_primary_ctrl_fd_rdhup: EPOLLRDHUP event handler.
|
||
* Closes primary_ctrl_fd, transitions to CTRL_ELECTING, truncates WAL to
|
||
* last_known_committed_lsn, fires ha_start_election_async (via cluster when non-NULL).
|
||
* Pass c == NULL to skip persist/election (unit test mode).
|
||
*/
|
||
void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c);
|
||
|
||
/*
|
||
* ha_arm_heartbeat_timeout_vtask: start/re-arm the heartbeat watchdog thread.
|
||
* Fires ha_on_primary_ctrl_fd_rdhup after election_timeout_ms + rand_jitter of
|
||
* silence. Calling ha_on_heartbeat_received() resets the deadline.
|
||
*/
|
||
void ha_arm_heartbeat_timeout_vtask(ha_state_t *ha, cluster_t *c);
|
||
|
||
/* ha_on_heartbeat_received: reset the heartbeat deadline (Req 6.3). */
|
||
void ha_on_heartbeat_received(ha_state_t *ha);
|
||
|
||
/* ha_disarm_heartbeat_timeout_vtask: stop the heartbeat watchdog thread. */
|
||
void ha_disarm_heartbeat_timeout_vtask(ha_state_t *ha);
|
||
|
||
/*
|
||
* ha_start_primary_heartbeat: arm a repeating delaymgr task that sends
|
||
* OPCode_nop to every standby in ha->standby_endpoints every
|
||
* heartbeat_interval_ms. Keeps the standby-side watchdog from timing out
|
||
* when the cluster is idle (no WAL entries, no config pushes).
|
||
* Must be called after ha->role is set to CTRL_PRIMARY.
|
||
* No-op when c == NULL (test mode).
|
||
*/
|
||
void ha_start_primary_heartbeat(cluster_t *c);
|
||
|
||
/* ── Task 2.4: Non-blocking async election initiator ────────────────────── */
|
||
|
||
/*
|
||
* ha_start_election_async: initiate a quorum election without blocking.
|
||
*
|
||
* Steps:
|
||
* 1. Store n_peers in ha->n_peers.
|
||
* 2. For single-node (n_peers == 1): call ha_elect_evaluate_and_signal directly
|
||
* with a self-only ha_elect_ctx_t — shortcircuit, no iomgr/delaymgr needed.
|
||
* 3. For multi-node: for each peer in peers[], allocate vote_ep_ctx_t, init
|
||
* endpoint with client_ssl_ctx, set on_vote_connected + vote_ep_err_handler,
|
||
* call endpoint_connect + iomgr_monitor; skip peer on connect failure.
|
||
* Allocate election_timeout_ctx_t (cancelled=0) and schedule
|
||
* delay_task_ms(delaymgr, election_timeout_ms, 1, election_timeout_fn, ctx).
|
||
* 4. Returns immediately; election result delivered via ha_elect_evaluate_and_signal.
|
||
*
|
||
* Observable: returns in < 5ms; endpoints registered in iomgr; watchdog in delaymgr.
|
||
* Requirements: 1.3 (non-blocking), 2.1 (single-node shortcircuit), 2.4 (async vote)
|
||
* Boundary: ha.c (election orchestration)
|
||
*/
|
||
void ha_start_election_async(ha_state_t *ha, cluster_t *c,
|
||
int n_peers, xce_candidate_t **peers,
|
||
uint64_t my_config_tag, uint64_t my_lsn,
|
||
const char *my_name);
|
||
|
||
/*
|
||
* ha_repl_ack_received: called when a standby ACKs config+WAL catch-up
|
||
* (OPCode_replCatchUpDone). Increments repl_ack_count; when majority
|
||
* threshold is reached, calls ha_promote_to_primary (Req 8.2).
|
||
*/
|
||
void ha_repl_ack_received(ha_state_t *ha, cluster_t *c);
|
||
|
||
/*
|
||
* ha_dispatch_repl_action_for_standby: evaluate the replication decision for
|
||
* one standby and execute the resulting action.
|
||
*
|
||
* Reads ha->standby_reported_tag[peer_idx] and ha->standby_reported_lsn[peer_idx]
|
||
* (stored by ha_process_standby_state on the previous call), calls
|
||
* ha_evaluate_repl_action, and acts on the result:
|
||
* - HA_REPL_PUSH_CONFIG: In production builds, calls push_config_to_standbys(c, …).
|
||
* In XCE_TESTING builds, calls ha_repl_push_config_for_standby_hook(c, peer_idx).
|
||
* - HA_REPL_STREAM_WAL: calls ha_stream_wal_to_standby(ep, wal_path, standby_lsn,
|
||
* committed_lsn) to batch-stream WAL entries via xce_request() (non-blocking).
|
||
* In XCE_TESTING builds, calls ha_repl_stream_wal_hook if non-NULL.
|
||
* - HA_REPL_SEND_SNAPSHOT: calls ha_send_snapshot_to_standby(ep, snap_path, lsn).
|
||
* In XCE_TESTING builds, calls ha_repl_send_snapshot_hook(peer_idx) if non-NULL.
|
||
* - HA_REPL_CONVERGED: no-op.
|
||
*
|
||
* One action per heartbeat cycle per standby (design §Unified Heartbeat-Driven
|
||
* Protocol). CONVERGED is a no-op.
|
||
*
|
||
* ep — outbound endpoint to the standby (used for STREAM_WAL; may be
|
||
* NULL in tests that only exercise PUSH_CONFIG or CONVERGED paths)
|
||
* peer_idx — 0-based index into ha->standby_reported_* arrays
|
||
* committed_tag — primary's committed config_tag (from c->myconf or test fixture)
|
||
* committed_lsn — primary's committed WAL LSN (from c->persist or test fixture)
|
||
* oldest_wal_lsn— oldest retained WAL LSN (0 = all WAL retained, skip snapshot)
|
||
*
|
||
* Returns 0 on success; negative on internal error.
|
||
* Requirements: 1.4, 5.1–5.5 (ha-repl-hardening spec)
|
||
*/
|
||
int ha_dispatch_repl_action_for_standby(ha_state_t *ha,
|
||
cluster_t *c,
|
||
endpoint_t *ep,
|
||
int peer_idx,
|
||
uint64_t committed_tag,
|
||
uint64_t committed_lsn,
|
||
uint64_t oldest_wal_lsn);
|
||
|
||
/*
|
||
* ha_send_ctrlhello_to_all_standbys: iterate ha->standby_endpoints and send
|
||
* an OP_hello CtrlHelloRequest to each connected standby.
|
||
* This triggers the existing catch-up replication path in handle_ctrlhello
|
||
* (Req 7.2). Defined in cluster.c; called from ha_elect_evaluate_and_signal
|
||
* in the COMMITTING path.
|
||
*/
|
||
void ha_send_ctrlhello_to_all_standbys(cluster_t *c);
|
||
|
||
/* ── P3-6: Test-only failover shim ────────────────────────────── */
|
||
|
||
#ifdef XCE_TESTING
|
||
|
||
/*
|
||
* ha_phase2_hook: test-only seam for cluster_init_phase2.
|
||
* When non-NULL, ha_elect_evaluate_and_signal() calls this INSTEAD OF the
|
||
* real cluster_init_phase2() so unit tests can count phase2 calls without
|
||
* requiring a real cluster_t. Pattern mirrors xce_vote_request_hook.
|
||
*/
|
||
extern int (*ha_phase2_hook)(void *c);
|
||
|
||
/*
|
||
* ha_send_ctrlhello_hook: test-only seam for ha_send_ctrlhello_to_all_standbys.
|
||
* When non-NULL, ha_elect_evaluate_and_signal() calls this in the COMMITTING
|
||
* path instead of the real function (cluster.c not linked in test binary).
|
||
* Requirements: 7.2.
|
||
*/
|
||
extern int (*ha_send_ctrlhello_hook)(void *c);
|
||
|
||
/*
|
||
* ha_push_config_hook: test-only seam for push_config_to_standbys.
|
||
* When non-NULL, ha_elect_evaluate_and_signal() calls this in the COMMITTING
|
||
* path instead of the real function. Requirements: 7.1.
|
||
*/
|
||
extern int (*ha_push_config_hook)(void *c);
|
||
|
||
/*
|
||
* ha_repl_push_config_for_standby_hook: test-only seam for the per-standby
|
||
* config push triggered by the heartbeat-reply decision loop (Task 3.3).
|
||
*
|
||
* When non-NULL, ha_dispatch_repl_action_for_standby() calls this INSTEAD OF
|
||
* push_config_to_standbys() when the decision is HA_REPL_PUSH_CONFIG, so unit
|
||
* tests can verify the wiring without requiring a real cluster_t or network.
|
||
*
|
||
* Signature: int hook(void *c, int peer_idx)
|
||
* c — cluster_t* (may be NULL in test mode)
|
||
* peer_idx — 0-based index of the standby that triggered the push
|
||
* Requirements: 1.4 (ha-repl-hardening)
|
||
*/
|
||
extern int (*ha_repl_push_config_for_standby_hook)(void *c, int peer_idx);
|
||
|
||
/*
|
||
* ha_repl_stream_wal_hook: test-only seam for ha_stream_wal_to_standby.
|
||
*
|
||
* When non-NULL, ha_dispatch_repl_action_for_standby() calls this INSTEAD OF
|
||
* ha_stream_wal_to_standby() when the decision is HA_REPL_STREAM_WAL, so unit
|
||
* tests can count batch-send invocations without requiring a real endpoint or
|
||
* WAL directory.
|
||
*
|
||
* Signature: int hook(void *ep, const char *wal_path,
|
||
* uint64_t standby_lsn, uint64_t committed_lsn)
|
||
* ep — endpoint_t* (may be NULL in test mode)
|
||
* wal_path — WAL directory path
|
||
* standby_lsn — standby's last applied LSN
|
||
* committed_lsn— primary's committed LSN
|
||
* Requirements: 5.1–5.5 (ha-repl-hardening)
|
||
*/
|
||
extern int (*ha_repl_stream_wal_hook)(void *ep, const char *wal_path,
|
||
uint64_t standby_lsn,
|
||
uint64_t committed_lsn);
|
||
|
||
/*
|
||
* ha_repl_send_snapshot_hook: test-only seam for ha_send_snapshot_to_standby.
|
||
*
|
||
* When non-NULL, ha_dispatch_repl_action_for_standby() calls this INSTEAD OF
|
||
* ha_send_snapshot_to_standby() when the decision is HA_REPL_SEND_SNAPSHOT,
|
||
* so unit tests can verify the wiring without requiring a real endpoint or
|
||
* snapshot directory.
|
||
*
|
||
* Signature: int hook(int peer_idx)
|
||
* peer_idx — 0-based index of the standby that triggered the snapshot
|
||
* Requirements: 4.1–4.5 (ha-repl-hardening)
|
||
*/
|
||
extern int (*ha_repl_send_snapshot_hook)(int peer_idx);
|
||
|
||
/*
|
||
* ha_tcp_connect_hook: test-only seam for the raw TCP connect inside
|
||
* ha_find_and_connect_to_primary. When non-NULL, the function calls this
|
||
* INSTEAD OF the real getaddrinfo/connect path, allowing unit tests to inject
|
||
* socketpair fds without real network I/O.
|
||
* Signature: int hook(const char *host, int port)
|
||
* Returns fd >= 0 on success, or -errno on failure.
|
||
* Requirements: task 9.2.
|
||
*/
|
||
extern int (*ha_tcp_connect_hook)(const char *host, int port);
|
||
|
||
/* 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_rdhup_election_hook: test-only seam for ha_start_election_async in the
|
||
* RDHUP failover path.
|
||
*
|
||
* When non-NULL, ha_on_primary_ctrl_fd_rdhup() calls this INSTEAD OF the real
|
||
* ha_start_election_async() so unit tests can verify the RDHUP path triggers
|
||
* async re-election without requiring a live iomgr/delaymgr.
|
||
*
|
||
* The hook receives the same arguments that would be passed to ha_start_election_async.
|
||
* Requirements: 6.3, 6.5, 9.3.
|
||
*/
|
||
extern void (*ha_rdhup_election_hook)(ha_state_t *ha, cluster_t *c,
|
||
int n_peers, xce_candidate_t **peers,
|
||
uint64_t my_config_tag, uint64_t my_lsn,
|
||
const char *my_name);
|
||
|
||
/*
|
||
* vote_ep_ctx_set_cluster_for_test: set the cluster_t pointer inside the
|
||
* vote_ep_ctx_t stored at ep->data. Used by unit tests that need
|
||
* ha_repl_ack_received to call ha_promote_to_primary(c) when quorum is reached.
|
||
* The vote_ep_ctx_t struct is internal to ha.c; this helper avoids exposing it.
|
||
* Requirements: task 7.3 (ha-peer-connection-reuse).
|
||
*/
|
||
void vote_ep_ctx_set_cluster_for_test(struct endpoint_t *ep, cluster_t *c);
|
||
#endif
|
||
|
||
#endif /* HA_H */
|