wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

chore: checkpoint all prior work for clean state

Browse source 
Captures all uncommitted work from ha-config-bootstrap,
ha-heartbeat-liveness, and associated spec/tooling updates:

- src: ha.c, ha.h, cluster.c, config_repl.c, myapp.c, raft.c —
  heartbeat liveness, RDHUP integration, committing phase fixes
- tests: test_ha_role.c, test_config_repl.c, test_ha_rdhup_integration.c,
  test_p1_integration.c, test_p3_hardening.c, scheduler test
- specs: election-iomgr-refactor, ha-heartbeat-liveness,
  ha-peer-connection-reuse, raft-library (new); ha-config-bootstrap
  and election-iomgr-refactor task/research updates
- tooling: .wolf/ state, ha/data cluster fixtures, raft/raft.tla,
  src/build symlink, xce-build skill update
This commit is contained in:
Rongsong Shen 2026-06-04 13:04:16 +00:00
parent 15bd291bdd
commit e31673dd88
66 changed files with 122422 additions and 860 deletions
Download patch file Download diff file Expand all files Collapse all files

74
src/controller/CLAUDE.md
View file

@ -25,6 +25,70 @@ Links against `libxce` from `src/lib/`.
- Component-specific headers in `include/`
- Shared types defined in `src/include/xce/` (libxce)
## HA Protocol Verification
### Raft TLA+ Formal Specification
The file `raft/raft.tla` contains Diego Ongaro's TLA+ specification of the Raft consensus algorithm. This is the authoritative reference for HA protocol correctness. **All HA protocol changes must be verified against this specification.**
### TLA+ Variable-to-Code Mapping
| TLA+ Variable | XCE Code | Location |
|--------------|----------|----------|
| `currentTerm[i]` | `ha->term` (_Atomic uint64_t) | `ha.h:139` |
| `state[i]` ∈ {Follower,Candidate,Leader} | `ha->role` ∈ {INIT,ELECTING,PRIMARY_COMMITTING,PRIMARY,STANDBY} | `ha.h:139` |
| `votedFor[i]` | `ha->voted_for_name[XCE_MAXNAME_LEN]` | `ha.h:161` |
| `log[i]` | WAL segments in `persist->wal_path` | `persist.c` |
| `commitIndex[i]` | `persist->committed_lsn` (_Atomic uint64_t) | `persist.h` |
| `lastApplied[i]` | `ha->applied_lsn` (_Atomic uint64_t) | `ha.h:141` |
| `nextIndex[i,j]` | `standby.join_applied_lsn` (replicator) | `repl.h` |
| `matchIndex[i,j]` | `standby.acked_lsn` (replicator) | `repl.h` |
### TLA+ Action-to-Code Mapping
| TLA+ Action | XCE Code | Location |
|------------|----------|----------|
| `Timeout(i)` | `raft_start_election` via delaymgr | `raft.c:57` |
| `RequestVote(i,j)` | `raft_on_vote_request` / reply path | `raft.c:106,169` |
| `BecomeLeader(i)` | `raft_cb_become_leader` | `ha.c:347` |
| `AdvanceCommitIndex(i)` | `repl_wait_majority_ack` → advance `committed_lsn` | `repl.c:338` |
| `AppendEntries(i,j)` | `OPCode_nop` heartbeat + WAL streaming | `ha.c:2880`, `config_repl.c:1489` |
| `ClientRequest(i,v)` | Job submit → WAL append → `repl_wait_majority_ack` | `persist.c`, `repl.c` |
| `UpdateTerm(i,j,m)` | Term CAS in `handle_request_vote` / vote reply handler | `ha.c:1218-1230` |
### Safety Invariants Verified
**Election Safety** (≤1 leader per term): Enforced by `votedFor` — a server grants at most one vote per term (`raft_on_vote_request` line 145-151). A candidate needs majority votes; each server votes at most once per term.
**Leader Completeness** (committed entries survive): Freshness gate in `RequestVote` — candidate must have `lastLogTerm ≥ voter` AND `lastLogIndex ≥ voter`. Since committed entries are on a majority, the new leader must have all committed entries. `persist_truncate_wal_above(safe_committed_lsn)` removes any divergent entries above the safe commit point.
**State Machine Safety** (no divergent entries): Monotonic `committed_lsn` + truncation on promotion. Entries at or below `committed_lsn` are identical across all servers.
**Commit Rule** (current-term entry required): Each WAL entry is committed individually via `repl_wait_majority_ack()`. Since entries are single-writer (only the current primary writes), every committed entry is implicitly from the current leader's term. Equivalent to Raft's batched commit rule.
### Post-Election Safe Commit Point (Critical)
After winning election, the new primary determines `safe_committed_lsn` from voter state collected during the election. This eliminates a round-trip vs waiting for the first heartbeat reply:
```
safe_committed_lsn = max(lsn : count of voters with peerLastLogLsn >= lsn) + 1 >= majority)
```
The `RequestVoteReply` message already carries `peerConfigTag` and `peerLastLogLsn` from each voter. The winning candidate uses these (from `ha_elect_ctx_t.votes[]`) to compute `safe_committed_lsn` immediately in `raft_cb_become_leader`, then calls `persist_truncate_wal_above(safe_committed_lsn)`.
**Never** use `last_known_committed_lsn` from the previous primary's ctrlhello for truncation — it may be stale (the old primary may have committed entries after the last ctrlhello was sent).
### Protocol Verification Checklist
When modifying HA protocol code (election, replication, heartbeat), verify:
- [ ] `votedFor` is updated atomically with term increment before granting any vote
- [ ] `currentTerm` is monotonic — never decreases
- [ ] `commitIndex` never decreases (monotonic)
- [ ] A leader only commits entries from its current term (or equivalent via per-entry majority ack)
- [ ] `safe_committed_lsn` is derived from voter replies, not from stale ctrlhello
- [ ] WAL truncation on promotion only removes entries above the safe commit point
- [ ] Freshness gate in RequestVote: candidate must have `lastLogTerm >= voter` AND `lastLogIndex >= voter`
## HA Design Constraints
### Inbound vs. Outbound Connection Ownership (CRITICAL)
@ -51,10 +115,12 @@ All inter-controller communication uses iomgr (epoll-based) and vtaskmgr:
- Blocking I/O (e.g., `fetch_config_from_primary`, `ha_find_and_connect_to_primary`) is only permitted from `delaymgr` worker threads, **never** from iomgr/vtask context.
### Heartbeat Model
XCE uses an asymmetric heartbeat model — different from standard Raft:
- The **standby** arms a watchdog timer (`ha_arm_heartbeat_timeout_vtask`) when it enters `CTRL_STANDBY`. If no message arrives from the primary within `heartbeat_timeout_ms`, it triggers re-election.
- The **primary does not send periodic heartbeats** to standbys. Instead, WAL replication entries, config pushes, and ctrlhello messages serve as liveness signals.
- `raft.c` (the election library) has no heartbeat loop — it is election-only. Do not add heartbeat callbacks to `raft_cbs_t`; that belongs in the HA layer.
XCE uses a bidirectional heartbeat model aligned with Raft's AppendEntries:
- The **primary** sends periodic heartbeats (`OPCode_nop`) to all standbys via `ha_start_primary_heartbeat` (delaymgr task, interval=3s). During `CTRL_PRIMARY_COMMITTING`, this also serves as the replication driver — the primary evaluates standby state from heartbeat replies and pushes config/WAL/snapshot as needed.
- The **standby** replies to each heartbeat with its current `(applied_lsn, config_tag)` in `RequestReply.stateReport`. This gives the primary continuous visibility into standby convergence.
- The **standby** also arms a liveness watchdog (`ha_arm_heartbeat_timeout_vtask`) when entering `CTRL_STANDBY`. If no message arrives from the primary within `heartbeat_timeout_ms` (10s), it triggers re-election via `ha_on_primary_ctrl_fd_rdhup`.
- `raft.c` (the election library) is election-only. Heartbeat and replication logic belongs in the HA layer (`ha.c`, `config_repl.c`).
### Connection Ownership
- Promoted vote connections (`ha->standby_endpoints`): owned by `vote_ep_err_handler`; freed when the standby disconnects.

12
src/controller/cluster.c
View file

@ -845,7 +845,7 @@ int cluster_init_phase1(cluster_t* c)
}
}
ha_install_sigusr1(&c->ha);
/* ha_install_sigusr1() removed — failover_requested replaced by ha_arm_heartbeat_timeout_vtask */
/* Initialize standby endpoint tracking */
c->ha.standby_endpoints = (zarray_t *)calloc(1, sizeof(zarray_t));
@ -1012,7 +1012,7 @@ int cluster_init_phase2(cluster_t* c)
}
}
/* Determine initial role based on am_i_primary() result set by myapp_run */
/* Role set by ha_elect_evaluate_and_signal() inside cluster_run() */
if (ha_is_primary(&c->ha))
ha_promote_to_primary(c);
else
@ -1257,12 +1257,8 @@ int cluster_start(cluster_t* c)
return rc;
}
/* Start HA heartbeat watchdog thread — P1-5 */
rc = ha_start_watchdog(&c->ha);
if (rc != 0) {
ERROR("ha_start_watchdog failed: %d", rc);
return rc;
}
/* ha_start_watchdog() removed — watchdog pthread replaced by
* ha_arm_heartbeat_timeout_vtask (delaymgr) */
delay_task(&(c->delaymgr), c->myconf->cluster_config->heartbeat_interval,
-1, (delay_task_entry_fn_t)check_workers_timeout, (void*)c);

1
src/controller/config_repl.c
View file

@ -1377,6 +1377,7 @@ int handle_ctrlhello(cluster_t *c, endpoint_t *ep, xce_request_t *req)
c->ha.primary_ctrl_fd == -1) {
c->ha.primary_ctrl_fd = ep->fd;
ha_on_heartbeat_received(&c->ha);
ha_arm_heartbeat_timeout_vtask(&c->ha, c);
INFO("HA: primary control channel established (fd=%d)", ep->fd);
}

397
src/controller/ha.c
View file

@ -441,17 +441,27 @@ static void raft_cb_become_leader(void *ctx, uint64_t term)
ha_start_primary_heartbeat(c);
/* Arm committing timeout watchdog */
{
/* Cancel any prior committing timeout (e.g. from a previous term) */
committing_timeout_ctx_t *prev = (committing_timeout_ctx_t *)
atomic_exchange(&ha->committing_timeout_ctx, NULL);
if (prev) {
atomic_store(&prev->cancelled, 1);
}
}
committing_timeout_ctx_t *ct = calloc(1, sizeof(*ct));
if (ct) {
atomic_init(&ct->cancelled, 0);
ct->ha = ha;
ct->c = c;
atomic_store(&ha->committing_timeout_ctx, ct);
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);
atomic_store(&ha->committing_timeout_ctx, NULL);
free(ct);
}
}
@ -474,6 +484,18 @@ static void raft_cb_become_follower(void *ctx, uint64_t term, int leader_idx)
/* Cancel any pending election timeout */
raft_cb_cancel_timeout(ctx);
/* Cancel any pending committing timeout — we are no longer leader.
* Without this, a node that stepped down due to a higher-term vote
* request would still promote to PRIMARY when the old committing
* timeout fires → split-brain. */
{
committing_timeout_ctx_t *ct = (committing_timeout_ctx_t *)
atomic_exchange(&ha->committing_timeout_ctx, NULL);
if (ct) {
atomic_store(&ct->cancelled, 1);
}
}
/* Sync ha->term */
atomic_store_explicit(&ha->term, term, memory_order_release);
@ -647,8 +669,7 @@ int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority)
atomic_store_explicit(&ha->promotion_failed, 0, memory_order_relaxed);
atomic_store_explicit(&ha->promotion_retry_count, 0, memory_order_relaxed);
atomic_store_explicit(&ha->in_catch_up, 0, memory_order_relaxed);
atomic_store_explicit(&ha->failover_requested, 0, memory_order_relaxed);
atomic_store_explicit(&ha->watchdog_stop, 0, memory_order_relaxed);
/* failover_requested / watchdog_stop deleted — watchdog replaced by ha_arm_heartbeat_timeout_vtask */
pthread_mutex_init(&ha->lock, NULL);
@ -706,30 +727,8 @@ int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority)
return 0;
}
/* ── Promotion / demotion / SIGUSR1 ────────────────────────── */
/* Async-signal-safe: atomic store on int-width type (POSIX.1-2017 §2.4.3).
* Pointer set once before signals are enabled — no locking required. */
static ha_state_t *g_sigusr1_ha = NULL;
static void sigusr1_handler(int sig)
{
(void)sig;
if (g_sigusr1_ha)
atomic_store_explicit(&g_sigusr1_ha->failover_requested, 1,
memory_order_release);
}
void ha_install_sigusr1(ha_state_t *ha)
{
g_sigusr1_ha = ha;
struct sigaction sa;
memset(&sa, 0, sizeof(sa));
sa.sa_handler = sigusr1_handler;
sigemptyset(&sa.sa_mask);
sa.sa_flags = SA_RESTART;
sigaction(SIGUSR1, &sa, NULL);
}
/* SIGUSR1 handler / ha_install_sigusr1() deleted — failover_requested replaced by
* ha_arm_heartbeat_timeout_vtask (delaymgr) in election-iomgr-refactor */
void ha_do_promote(ha_state_t *ha)
{
@ -754,182 +753,9 @@ void ha_do_demote(ha_state_t *ha)
atomic_load_explicit(&ha->term, memory_order_acquire));
}
/* ── P1-6: Election probing ─────────────────────────────────── */
/* probe_thread() / wait_any_probe() deleted — only called by am_i_primary() (election-iomgr-refactor) */
/*
* probe_thread: attempt a TCP (optionally SSL) connection to one candidate.
* Sets ctx->got_ack = 1 if the connection succeeds before election_done is set.
*/
static void *probe_thread(void *arg)
{
probe_ctx_t *ctx = (probe_ctx_t *)arg;
struct addrinfo hints, *res = NULL;
memset(&hints, 0, sizeof(hints));
hints.ai_family = AF_UNSPEC;
hints.ai_socktype = SOCK_STREAM;
char port_str[16];
snprintf(port_str, sizeof(port_str), "%u", (unsigned)ctx->candidate->port);
if (getaddrinfo(ctx->candidate->hostname, port_str, &hints, &res) != 0)
goto done;
int sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (sock < 0) { freeaddrinfo(res); goto done; }
/* Non-blocking connect */
int flags = fcntl(sock, F_GETFL, 0);
fcntl(sock, F_SETFL, flags | O_NONBLOCK);
connect(sock, res->ai_addr, res->ai_addrlen);
freeaddrinfo(res);
res = NULL;
/* Poll loop: check election_done between iterations */
while (!atomic_load(ctx->election_done)) {
struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP };
int pr = poll(&pfd, 1, (int)ctx->timeout_ms);
if (pr < 0 && errno == EINTR) continue;
if (pr == 0) goto cleanup; /* poll timeout */
if (pfd.revents & (POLLERR | POLLHUP)) goto cleanup;
if (pfd.revents & POLLOUT) break;
}
if (atomic_load(ctx->election_done)) goto cleanup;
/* Verify connect succeeded */
{
int err = 0;
socklen_t el = sizeof(err);
getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
if (err != 0) goto cleanup;
}
/* SSL handshake — only if ssl_ctx is set */
if (ctx->ssl_ctx) {
SSL *ssl = SSL_new((SSL_CTX *)ctx->ssl_ctx);
if (!ssl) goto cleanup;
/* Restore blocking for SSL */
fcntl(sock, F_SETFL, flags & ~O_NONBLOCK);
SSL_set_fd(ssl, sock);
int sr;
do {
sr = SSL_connect(ssl);
if (sr <= 0) {
int ssl_err = SSL_get_error(ssl, sr);
if (ssl_err != SSL_ERROR_WANT_READ &&
ssl_err != SSL_ERROR_WANT_WRITE)
break;
}
} while (sr <= 0 && !atomic_load(ctx->election_done));
SSL_shutdown(ssl);
SSL_free(ssl);
if (sr <= 0 || atomic_load(ctx->election_done))
goto cleanup;
}
ctx->got_ack = 1;
cleanup:
close(sock);
done:
return NULL;
}
/*
* wait_any_probe: spin-poll until any probe sets got_ack, or deadline passes.
* Returns 1 if any probe acknowledged, 0 if deadline expired.
*/
static int wait_any_probe(probe_ctx_t *probes, int n,
const struct timespec *deadline)
{
while (1) {
for (int i = 0; i < n; i++) {
if (probes[i].got_ack)
return 1;
}
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
if (now.tv_sec > deadline->tv_sec ||
(now.tv_sec == deadline->tv_sec &&
now.tv_nsec >= deadline->tv_nsec))
return 0;
/* Sleep 10ms between checks */
struct timespec sleep_ts = { .tv_sec = 0, .tv_nsec = 10 * 1000000L };
nanosleep(&sleep_ts, NULL);
}
}
/*
* am_i_primary: election probe logic.
*
* Priority-0 is unconditionally primary — never probes.
* Priority-N probes the N candidates with lower index (higher priority).
* If any responds before election_timeout + priority_step*N + jitter, this
* node becomes standby. If deadline expires with no response, self-promote.
*/
int am_i_primary(cluster_t *c)
{
/* Priority-0 never probes — it is unconditionally primary */
if (c->ha.my_priority == 0)
return 1;
const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
int n_probes = (c->ha.my_priority < pcfg->candidates_count)
? c->ha.my_priority
: pcfg->candidates_count;
if (n_probes == 0)
return 1;
/* Seed jitter PRNG if needed */
if (c->ha.jitter_seed == 0)
ha_seed_jitter(&c->ha.jitter_seed, c->ha.my_priority);
atomic_int election_done;
atomic_init(&election_done, 0);
probe_ctx_t probes[MAX_HA_CANDIDATES];
for (int i = 0; i < n_probes; i++) {
probes[i].candidate = (xce_candidate_t *)&pcfg->candidates[i];
probes[i].ssl_ctx = c->ha.ssl_ctx;
probes[i].timeout_ms = c->ha.config.probe_connect_timeout_ms;
probes[i].election_done = &election_done;
probes[i].got_ack = 0;
pthread_create(&probes[i].thread, NULL, probe_thread, &probes[i]);
}
/* Compute deadline = election_timeout + priority_step * my_priority + jitter */
long to_ms = (long)c->ha.config.election_timeout_ms
+ (long)c->ha.my_priority
* (long)c->ha.config.election_priority_step_ms;
if (c->ha.config.election_jitter_max_ms > 0) {
to_ms += (long)(xorshift64_next(&c->ha.jitter_seed)
% ((uint64_t)c->ha.config.election_jitter_max_ms + 1));
}
struct timespec deadline;
clock_gettime(CLOCK_MONOTONIC, &deadline);
deadline.tv_sec += to_ms / 1000;
deadline.tv_nsec += (to_ms % 1000) * 1000000L;
if (deadline.tv_nsec >= 1000000000L) {
deadline.tv_sec++;
deadline.tv_nsec -= 1000000000L;
}
int got_ack = wait_any_probe(probes, n_probes, &deadline);
/* Signal all threads to stop */
atomic_store(&election_done, 1);
for (int i = 0; i < n_probes; i++)
pthread_join(probes[i].thread, NULL);
if (got_ack) {
WARN("HA election: higher-priority candidate responded — "
"starting as standby (my_priority=%d)", c->ha.my_priority);
return 0;
}
INFO("HA election: no higher-priority candidate responded within %ldms — "
"starting as primary (my_priority=%d)", to_ms, c->ha.my_priority);
return 1;
}
/* am_i_primary() deleted — replaced by ha_start_election_async (election-iomgr-refactor) */
/*
* ha_is_primary: lock-free role check.
@ -941,23 +767,7 @@ 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:
* Stub implementations — full listener is implemented in Task 7 (cluster.c).
* The stubs allow am_i_primary() callers (main.c / cluster.c) to compile and
* link without requiring the full Phase-2 network stack.
*/
void start_heartbeat_listener(cluster_t *c)
{
(void)c;
/* Heartbeat listener stub — full implementation in cluster.c Task 7 */
}
void stop_heartbeat_listener(cluster_t *c)
{
(void)c;
}
/* start_heartbeat_listener() / stop_heartbeat_listener() deleted — never called (election-iomgr-refactor) */
/* ── P3-6: Test-only failover shim ────────────────────────────── */
@ -970,96 +780,14 @@ uint64_t ha_trigger_manual_failover_for_test(ha_state_t *ha)
}
#endif
/* ha_check_heartbeat_nocluster() removed from the public API — logic inlined in
* ha_watchdog_thread(). XCE_TESTING shim preserved so existing unit tests compile. */
#ifdef XCE_TESTING
void ha_check_heartbeat_nocluster(ha_state_t *ha)
{
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
if (role != CTRL_STANDBY) return;
if (ha->primary_host[0] == '\0') return;
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
pthread_mutex_lock(&ha->lock);
struct timespec last = ha->last_heartbeat;
pthread_mutex_unlock(&ha->lock);
uint64_t elapsed_ms =
(uint64_t)(now.tv_sec - last.tv_sec) * 1000ULL +
(uint64_t)(now.tv_nsec - last.tv_nsec) / 1000000ULL;
if (elapsed_ms >= ha->config.heartbeat_timeout_ms)
atomic_store_explicit(&ha->failover_requested, 1, memory_order_release);
}
#endif /* XCE_TESTING */
static void *ha_watchdog_thread(void *arg)
{
ha_state_t *ha = arg;
while (!atomic_load_explicit(&ha->watchdog_stop, memory_order_acquire)) {
long ms = (long)ha->config.heartbeat_interval_ms;
struct timespec ts = {
.tv_sec = ms / 1000,
.tv_nsec = (ms % 1000) * 1000000L,
};
nanosleep(&ts, NULL);
/* Re-check stop flag after sleep */
if (atomic_load_explicit(&ha->watchdog_stop, memory_order_acquire)) break;
/* Heartbeat timeout check (inlined from deleted ha_check_heartbeat_nocluster) */
{
ctrl_role_t role =
(ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
if (role == CTRL_STANDBY && ha->primary_host[0] != '\0') {
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
pthread_mutex_lock(&ha->lock);
struct timespec last = ha->last_heartbeat;
pthread_mutex_unlock(&ha->lock);
uint64_t elapsed_ms =
(uint64_t)(now.tv_sec - last.tv_sec) * 1000ULL +
(uint64_t)(now.tv_nsec - last.tv_nsec) / 1000000ULL;
if (elapsed_ms >= ha->config.heartbeat_timeout_ms)
atomic_store_explicit(&ha->failover_requested, 1,
memory_order_release);
}
}
if (atomic_load_explicit(&ha->failover_requested, memory_order_acquire)) {
atomic_store_explicit(&ha->failover_requested, 0, memory_order_release);
/* Promotion requires cluster_t; the watchdog only has ha_state_t.
* Full ha_promote_to_primary(c) is called from the cluster's own
* failover path which holds cluster_t context. */
}
}
return NULL;
}
int ha_start_watchdog(ha_state_t *ha)
{
atomic_store_explicit(&ha->watchdog_stop, 0, memory_order_release);
int rc = pthread_create(&ha->watchdog_thread, NULL, ha_watchdog_thread, ha);
if (rc != 0) return -rc;
return 0;
}
void ha_stop_watchdog(ha_state_t *ha)
{
atomic_store_explicit(&ha->watchdog_stop, 1, memory_order_release);
if (ha->watchdog_thread != 0) {
pthread_join(ha->watchdog_thread, NULL);
ha->watchdog_thread = 0;
}
}
/* ha_check_heartbeat_nocluster() / ha_watchdog_thread() / ha_start_watchdog() /
* ha_stop_watchdog() deleted — watchdog pthread replaced by
* ha_arm_heartbeat_timeout_vtask (delaymgr) in election-iomgr-refactor */
void ha_fini(ha_state_t *ha)
{
if (!ha)
return;
ha_stop_watchdog(ha); /* join watchdog thread before destroying mutex */
pthread_mutex_destroy(&ha->lock);
sem_destroy(&ha->heartbeat_seen);
@ -1473,10 +1201,42 @@ int handle_request_vote(ha_state_t *ha,
uint64_t my_last_lsn,
int in_catch_up_flag)
{
/* Req 6.1, 6.2: role-based deny gate — checked before any Raft state mutation.
* PRIMARY and PRIMARY_COMMITTING never grant votes (protect the live primary).
* STANDBY with a fresh heartbeat also denies (primary is alive, no new election
* needed). CTRL_INIT and CTRL_ELECTING fall through to Raft evaluation. */
/* Req 2.3: higher term → step down BEFORE any role gate.
*
* Raft §5.1: "If RPC request or response contains term T > currentTerm:
* set currentTerm = T, convert to follower."
*
* Without this ordering, a node in CTRL_PRIMARY_COMMITTING at term=N
* would deny a vote request at term=N+1 without ever seeing the higher
* term. Both nodes then independently promote to PRIMARY via their
* committing timeout → split-brain. */
uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire);
if (term > my_term) {
atomic_store_explicit(&ha->term, term, memory_order_release);
ha->voted_for_term = 0;
memset(ha->voted_for_name, 0, sizeof(ha->voted_for_name));
/* Step down from leader/committing: a higher-term election is in
* progress and we must not block it with the role gate below. */
ctrl_role_t cur_role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
if (cur_role == CTRL_PRIMARY || cur_role == CTRL_PRIMARY_COMMITTING) {
ha_do_demote(ha);
INFO("HA: stepping down (term %" PRIu64 " → %" PRIu64
") — higher-term election in progress", my_term, term);
}
my_term = term;
}
/* Req 2.2: stale term → deny, do not update any state */
if (term < my_term) {
return 0;
}
/* Req 6.1, 6.2: role-based deny gate.
* Evaluated AFTER the term catch-up above: if we just stepped down due to
* a higher term, role is now CTRL_STANDBY and we fall through to voting.
* PRIMARY and PRIMARY_COMMITTING at the current term still deny votes
* (protecting the live primary from same-term challengers). */
ctrl_role_t my_role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
if (my_role == CTRL_PRIMARY || my_role == CTRL_PRIMARY_COMMITTING)
@ -1491,21 +1251,6 @@ int handle_request_vote(ha_state_t *ha,
/* Protect voted_for check+set from concurrent vote requests on different vtasks */
pthread_mutex_lock(&ha->lock);
uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire);
/* Req 2.2: stale term → deny, do not update any state */
if (term < my_term) {
pthread_mutex_unlock(&ha->lock);
return 0;
}
/* Req 2.3: higher term → advance term, clear prior vote record */
if (term > my_term) {
atomic_store_explicit(&ha->term, term, memory_order_release);
ha->voted_for_term = 0;
memset(ha->voted_for_name, 0, sizeof(ha->voted_for_name));
}
/* Req 2.6: already voted for a different candidate in this term → deny */
if (ha->voted_for_term == term &&
ha->voted_for_name[0] != '\0' &&
@ -1737,17 +1482,25 @@ apply_role:
/* Arm committing timeout watchdog (degraded-mode fallback, Req 8.4) */
if (c) {
/* Cancel any prior committing timeout */
committing_timeout_ctx_t *prev = (committing_timeout_ctx_t *)
atomic_exchange(&ha->committing_timeout_ctx, NULL);
if (prev) {
atomic_store(&prev->cancelled, 1);
}
committing_timeout_ctx_t *ct = calloc(1, sizeof(*ct));
if (ct) {
atomic_init(&ct->cancelled, 0);
ct->ha = ha;
ct->c = c;
atomic_store(&ha->committing_timeout_ctx, ct);
int rc = delay_task_ms(&c->delaymgr,
(int)ha->config.committing_timeout_ms,
1, committing_timeout_fn, ct);
if (rc < 0) {
WARN("ha_elect_evaluate_and_signal: delay_task_ms "
"for committing timeout failed: %d", rc);
atomic_store(&ha->committing_timeout_ctx, NULL);
free(ct);
}
}

64
src/controller/ha.h
View file

@ -131,14 +131,7 @@ typedef struct {
uint32_t committing_timeout_ms; /* Quorum-ACK deadline for CTRL_PRIMARY_COMMITTING (default 10000) */
} 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;
/* 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 {
@ -152,14 +145,12 @@ typedef struct {
_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 */
/* 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 thread — started in P1-3 */
_Atomic int watchdog_stop;
pthread_t watchdog_thread;
/* 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 */
@ -183,6 +174,7 @@ typedef struct {
/* ── 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 */
} ha_state_t;
@ -201,11 +193,7 @@ int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority);
*/
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);
/* am_i_primary() deleted — replaced by ha_start_election_async (election-iomgr-refactor) */
/*
* ha_promote_to_primary: stub — atomically sets role to PRIMARY.
@ -234,35 +222,6 @@ void ha_do_promote(ha_state_t *ha);
*/
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 ──────────────────────── */
@ -289,18 +248,7 @@ 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);
/* start_heartbeat_listener() / stop_heartbeat_listener() deleted — never called (election-iomgr-refactor) */
/*
* ha_dispatch_gate: role-based operation filter for the vtaskmgr dispatch loop.

35
src/controller/myapp.c
View file

@ -68,6 +68,12 @@ int myapp_run(myconf_t *myconf)
return rc;
}
/* Wire raft_c before iomgr starts so raft_cb_persist can look up
* candidate names when processing incoming vote requests during the
* startup stagger nap. Without this, voted_for_name is cleared and the
* already_voted check fails → unnecessary duplicate elections. */
c.ha.raft_c = &c;
/* ── Step 2: SSL contexts ─────────────────────────────────────────────── */
/* Validate cert/key files early to fail fast before any network setup */
@ -227,17 +233,30 @@ int myapp_run(myconf_t *myconf)
nanosleep(&jitter, NULL);
}
/* If we already voted for someone during the startup delay (incoming vote
* request arrived from the higher-priority node that started first), skip
* starting our own election — we'll become STANDBY via the election timeout
* watchdog on the winner's side, or promote ourselves if no heartbeat arrives. */
/* If a leader is already known after the startup stagger (either because
* iomgr processed an incoming vote request and updated the raft term, or
* because ctrlhello arrived on a promoted vote ep), skip starting our own
* election. The voted_for_name check catches the synchronous vote-grant
* path (iomgr was already running); the raft term + primary_ctrl_fd checks
* catch the case where iomgr was NOT running during the nap (connections
* queued in the kernel backlog) but the winner's ctrlhello has already
* established the control channel. */
{
int leader_known = 0;
pthread_mutex_lock(&c.ha.lock);
int already_voted = (c.ha.voted_for_name[0] != '\0');
if (c.ha.voted_for_name[0] != '\0')
leader_known = 1;
pthread_mutex_unlock(&c.ha.lock);
if (already_voted) {
INFO("myapp_run: already voted during startup delay — "
"skipping own election, waiting for primary");
if (!leader_known && c.ha.raft.term > 0)
leader_known = 1;
if (!leader_known && c.ha.primary_ctrl_fd >= 0)
leader_known = 1;
if (leader_known) {
INFO("myapp_run: leader already known after startup delay "
"(voted_for=%s, raft_term=%" PRIu64 ", primary_fd=%d) — "
"skipping own election",
c.ha.voted_for_name[0] ? c.ha.voted_for_name : "(none)",
c.ha.raft.term, c.ha.primary_ctrl_fd);
goto skip_election;
}
}

2
src/controller/testsuite/controller_test.c
View file

@ -20,7 +20,6 @@ extern void test_e2e_hooks(void);
extern void test_e2e_gc(void);
extern void test_e2e_capnp(void);
extern void test_ha_role(void);
extern void test_am_i_primary(void);
extern void test_ha_cluster(void);
extern void test_ha_redirect(void);
extern void test_p1_integration(void);
@ -140,7 +139,6 @@ int main(int argc, char* argv[])
TEST(test_e2e_gc);
TEST(test_e2e_capnp);
TEST(test_ha_role);
TEST(test_am_i_primary);
TEST(test_ha_cluster);
TEST(test_ha_redirect);
TEST(test_p1_integration);

5
src/controller/testsuite/test_config_repl.c
View file

@ -886,13 +886,14 @@ static void test_upgrade_config_req_extract_raw(void)
free(out_sha256);
free(buf);
/* The decoded-form variant must always return -ENOTSUP */
/* xce_upgrade_config_req_extract: returns -EINVAL on a zeroed request
* (body_which is 0, not Request_body_upgradeconfigrequestmsg). */
{
xce_request_t dummy_req;
memset(&dummy_req, 0, sizeof(dummy_req));
uint8_t *b = NULL, *s = NULL;
size_t l = 0;
TEST_ASSERT_EQUAL_INT(-ENOTSUP,
TEST_ASSERT_EQUAL_INT(-EINVAL,
xce_upgrade_config_req_extract(&dummy_req, &b, &l, &s));
}

440
src/controller/testsuite/test_ha_rdhup_integration.c Normal file
View file

@ -0,0 +1,440 @@
/*
* test_ha_rdhup_integration.c — Task 6.4 integration test: RDHUP failover
* end-to-end role transition.
*
* Tests verify the full RDHUP → CTRL_ELECTING → CTRL_PRIMARY_COMMITTING →
* CTRL_PRIMARY state-machine chain using XCE_TESTING hook seams. No live
* sockets or real cluster_t required.
*
* Requirements: 6.3, 6.5, 9.2, 9.3 (Req numbers from tasks.md 5.2 / 6.4)
*
* Hook seams used:
* ha_rdhup_election_hook — intercepts ha_start_election_async in RDHUP path
* ha_phase2_hook — stubs cluster_init_phase2 (already done = 1)
* ha_push_config_hook — stubs push_config_to_standbys
* ha_send_ctrlhello_hook — stubs ha_send_ctrlhello_to_all_standbys
*
* TDD: tests written first (RED); implementation verified GREEN.
*/
#define _GNU_SOURCE
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
#include <unity/unity.h>
#include "ha.h"
#include "myconf.h"
#include "xce/cluster_def.h"
/* ── XCE_TESTING hook seam externs ──────────────────────────────────────── */
extern int (*ha_phase2_hook)(void *c);
extern int (*ha_push_config_hook)(void *c);
extern int (*ha_send_ctrlhello_hook)(void *c);
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);
/* ── Stub counters ───────────────────────────────────────────────────────── */
static int g_phase2_calls = 0;
static int g_push_config_calls = 0;
static int g_ctrlhello_calls = 0;
static int stub_phase2(void *c) { (void)c; g_phase2_calls++; return 0; }
static int stub_push_config(void *c) { (void)c; g_push_config_calls++; return 0; }
static int stub_ctrlhello(void *c) { (void)c; g_ctrlhello_calls++; return 0; }
/* ── Shared HA config ───────────────────────────────────────────────────── */
static ha_config_t g_cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.election_jitter_max_ms = 0,
.election_priority_step_ms = 500,
.committing_timeout_ms = 10000,
};
/* ── install_stubs / clear_stubs ─────────────────────────────────────────── */
static void install_stubs(void)
{
g_phase2_calls = 0;
g_push_config_calls = 0;
g_ctrlhello_calls = 0;
ha_phase2_hook = stub_phase2;
ha_push_config_hook = stub_push_config;
ha_send_ctrlhello_hook = stub_ctrlhello;
}
static void clear_stubs(void)
{
ha_rdhup_election_hook = NULL;
ha_phase2_hook = NULL;
ha_push_config_hook = NULL;
ha_send_ctrlhello_hook = NULL;
}
/* ── Inline RDHUP election hook: drives evaluate immediately ─────────────
*
* Simulates ha_start_election_async completing synchronously:
* allocates a minimal ha_elect_ctx_t with self-vote won, calls
* ha_elect_evaluate_and_signal, then frees ctx.
*
* This models the single-node shortcircuit path (Req 9.1) adapted to
* a multi-node cluster where the surviving standby immediately wins —
* the key assertion is role progression, not peer negotiation.
*/
typedef struct {
cluster_t *c;
} rdhup_hook_state_t;
static rdhup_hook_state_t g_rdhup_state;
static void rdhup_election_drives_evaluate(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)
{
(void)peers; (void)my_config_tag; (void)my_lsn; (void)my_name;
/* Allocate minimal ctx with quorum already satisfied (self-vote wins) */
int n_other = (n_peers > 1) ? n_peers - 1 : 0;
ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) +
(size_t)n_other * sizeof(ctx->votes[0]));
if (!ctx) return;
ctx->n_peers = n_other;
ctx->votes_granted = 1; /* self-vote */
ctx->n_replied = n_other;
ctx->result = -1;
ctx->done_posted = 0;
ctx->c = g_rdhup_state.c;
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
ha_elect_evaluate_and_signal(ha, ctx);
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
free(ctx);
}
/* ── Helper: make a minimal 2-candidate cluster_t ───────────────────────── */
static void make_cluster(cluster_t *c, myconf_t *myconf, const char *my_name)
{
memset(c, 0, sizeof(*c));
memset(myconf, 0, sizeof(*myconf));
myconf->name = (char *)my_name;
myconf->peers_cfg.candidates_count = 2;
snprintf(myconf->peers_cfg.candidates[0].name,
sizeof(myconf->peers_cfg.candidates[0].name), "%s", my_name);
myconf->peers_cfg.candidates[0].port = 7001;
snprintf(myconf->peers_cfg.candidates[1].name,
sizeof(myconf->peers_cfg.candidates[1].name), "ctrl-b");
myconf->peers_cfg.candidates[1].port = 7002;
c->myconf = myconf;
}
/* ═══════════════════════════════════════════════════════════════════════════
* Test 1: RDHUP → ELECTING → COMMITTING (no standbys present → via hook chain)
*
* Req 6.3: RDHUP triggers CTRL_ELECTING + async election
* Req 9.3: RDHUP election uses ha_start_election_async (not am_i_primary_quorum)
* Req 9.2: phase2_done=1 suppresses cluster_init_phase2 on RDHUP re-election
* ═════════════════════════════════════════════════════════════════════════ */
static void test_rdhup_full_path_reaches_committing(void)
{
install_stubs();
cluster_t c;
myconf_t myconf;
make_cluster(&c, &myconf, "ctrl-a");
TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &g_cfg, 0));
ha_do_demote(&c.ha);
c.ha.phase2_done = 1; /* surviving standby — phase2 already done */
c.ha.n_peers = 1; /* 1 peer (2-node cluster) */
/* Wire the hook that drives evaluate immediately */
g_rdhup_state.c = &c;
ha_rdhup_election_hook = rdhup_election_drives_evaluate;
/* Give RDHUP handler a socketpair fd to close */
int sv[2];
TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv));
c.ha.primary_ctrl_fd = sv[0];
/* Fire RDHUP */
ha_on_primary_ctrl_fd_rdhup(&c.ha, &c);
clear_stubs();
close(sv[1]);
/* phase2 must NOT have been called (phase2_done=1 → skip) */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_phase2_calls,
"RDHUP re-election must not call cluster_init_phase2 when phase2_done=1 (Req 9.2)");
/* Role must be CTRL_PRIMARY_COMMITTING (quorum won → committing gate) */
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role,
"RDHUP full path must reach CTRL_PRIMARY_COMMITTING after winning election (Req 6.3)");
ha_fini(&c.ha);
printf("test_rdhup_full_path_reaches_committing: PASS (role=%d==CTRL_PRIMARY_COMMITTING)\n",
role);
}
/* ═══════════════════════════════════════════════════════════════════════════
* Test 2: RDHUP → COMMITTING → PRIMARY via ha_repl_ack_received
*
* Req 8.2: majority ack in COMMITTING → CTRL_PRIMARY
* Req 6.5: promoted standby serves traffic (dispatch gate opens)
* ═════════════════════════════════════════════════════════════════════════ */
static void test_rdhup_committing_to_primary_on_ack(void)
{
install_stubs();
cluster_t c;
myconf_t myconf;
make_cluster(&c, &myconf, "ctrl-a");
TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &g_cfg, 0));
ha_do_demote(&c.ha);
c.ha.phase2_done = 1;
c.ha.n_peers = 1; /* 2-node cluster: quorum=1, threshold=0 */
g_rdhup_state.c = &c;
ha_rdhup_election_hook = rdhup_election_drives_evaluate;
int sv[2];
TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv));
c.ha.primary_ctrl_fd = sv[0];
/* Fire RDHUP → should reach CTRL_PRIMARY_COMMITTING */
ha_on_primary_ctrl_fd_rdhup(&c.ha, &c);
close(sv[1]);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT((int)CTRL_PRIMARY_COMMITTING, role);
/* n_peers=1 → quorum=(1+2)/2=1 → threshold=0 → first ack promotes.
* ha_repl_ack_received increments count; acks(1) >= quorum-1(0) → promote. */
ha_repl_ack_received(&c.ha, &c);
role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"ha_repl_ack_received must drive COMMITTING → CTRL_PRIMARY on quorum ack (Req 8.2)");
clear_stubs();
ha_fini(&c.ha);
printf("test_rdhup_committing_to_primary_on_ack: PASS (role=%d==CTRL_PRIMARY)\n", role);
}
/* ═══════════════════════════════════════════════════════════════════════════
* Test 3: RDHUP in single-node cluster → immediate PRIMARY (no COMMITTING)
*
* Req 9.1: single-node self-elects without network
* n_peers=0 → quorum=1, threshold=0 → acks(0) >= 0 → promotes immediately
* in ha_elect_evaluate_and_signal (no peers to wait for acks from)
* ═════════════════════════════════════════════════════════════════════════ */
static void test_rdhup_single_node_reaches_primary(void)
{
install_stubs();
cluster_t c;
myconf_t myconf;
make_cluster(&c, &myconf, "ctrl-a");
/* Override to single-node */
myconf.peers_cfg.candidates_count = 1;
TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &g_cfg, 0));
ha_do_demote(&c.ha);
c.ha.phase2_done = 1;
c.ha.n_peers = 0; /* single node: no peers */
g_rdhup_state.c = &c;
ha_rdhup_election_hook = rdhup_election_drives_evaluate;
int sv[2];
TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv));
c.ha.primary_ctrl_fd = sv[0];
ha_on_primary_ctrl_fd_rdhup(&c.ha, &c);
close(sv[1]);
/* n_peers=0 → quorum=(0+2)/2=1 → threshold=0 → acks(0) >= 0 → promoted inline */
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
/* Single-node: no standbys to ack, committing_timeout degrades to PRIMARY.
* The committing timeout is 10s; for single-node we call ha_repl_ack_received
* once to simulate the degraded-mode zero-threshold promotion (threshold=0). */
if (role == (int)CTRL_PRIMARY_COMMITTING) {
ha_repl_ack_received(&c.ha, &c);
role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
}
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"single-node RDHUP path must reach CTRL_PRIMARY (Req 9.1)");
clear_stubs();
ha_fini(&c.ha);
printf("test_rdhup_single_node_reaches_primary: PASS (role=%d==CTRL_PRIMARY)\n", role);
}
/* ═══════════════════════════════════════════════════════════════════════════
* Test 4: 3-node cluster RDHUP — two acks needed, premature ack stays COMMITTING
*
* Req 8.2: acks < quorum-1 → stays COMMITTING; acks >= quorum-1 → PRIMARY
* 3-node (n_peers=2): quorum=2, threshold=1 → need 1 ack to promote
* ═════════════════════════════════════════════════════════════════════════ */
static void test_rdhup_3node_one_ack_promotes(void)
{
install_stubs();
cluster_t c;
myconf_t myconf;
make_cluster(&c, &myconf, "ctrl-a");
myconf.peers_cfg.candidates_count = 3;
snprintf(myconf.peers_cfg.candidates[2].name,
sizeof(myconf.peers_cfg.candidates[2].name), "ctrl-c");
myconf.peers_cfg.candidates[2].port = 7003;
TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &g_cfg, 0));
ha_do_demote(&c.ha);
c.ha.phase2_done = 1;
c.ha.n_peers = 2; /* 3-node: quorum=2, threshold=1 */
g_rdhup_state.c = &c;
ha_rdhup_election_hook = rdhup_election_drives_evaluate;
int sv[2];
TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv));
c.ha.primary_ctrl_fd = sv[0];
ha_on_primary_ctrl_fd_rdhup(&c.ha, &c);
close(sv[1]);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role,
"3-node RDHUP must enter CTRL_PRIMARY_COMMITTING before acks");
/* First ack: count=1 >= threshold=1 → promotes */
ha_repl_ack_received(&c.ha, &c);
role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role,
"3-node: first ack (count=1 >= threshold=1) must promote to CTRL_PRIMARY (Req 8.2)");
clear_stubs();
ha_fini(&c.ha);
printf("test_rdhup_3node_one_ack_promotes: PASS (role=%d==CTRL_PRIMARY)\n", role);
}
/* ═══════════════════════════════════════════════════════════════════════════
* Test 5: config push and ctrlhello hooks fire during COMMITTING entry
*
* Req 7.1: push_config_to_standbys called after winning election
* Req 7.2: ha_send_ctrlhello_to_all_standbys called after winning election
* ═════════════════════════════════════════════════════════════════════════ */
static void test_rdhup_committing_fires_replication_hooks(void)
{
install_stubs();
cluster_t c;
myconf_t myconf;
make_cluster(&c, &myconf, "ctrl-a");
TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &g_cfg, 0));
ha_do_demote(&c.ha);
c.ha.phase2_done = 1;
c.ha.n_peers = 1;
g_rdhup_state.c = &c;
ha_rdhup_election_hook = rdhup_election_drives_evaluate;
int sv[2];
TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv));
c.ha.primary_ctrl_fd = sv[0];
ha_on_primary_ctrl_fd_rdhup(&c.ha, &c);
close(sv[1]);
TEST_ASSERT_GREATER_THAN_INT_MESSAGE(0, g_push_config_calls,
"push_config_to_standbys hook must fire when entering COMMITTING (Req 7.1)");
TEST_ASSERT_GREATER_THAN_INT_MESSAGE(0, g_ctrlhello_calls,
"ha_send_ctrlhello hook must fire when entering COMMITTING (Req 7.2)");
clear_stubs();
ha_fini(&c.ha);
printf("test_rdhup_committing_fires_replication_hooks: PASS "
"(push=%d, ctrlhello=%d)\n",
g_push_config_calls, g_ctrlhello_calls);
}
/* ═══════════════════════════════════════════════════════════════════════════
* Test 6: dispatch gate opens after CTRL_PRIMARY (job ops allowed)
*
* Req 6.5: after RDHUP promotion, primary accepts job dispatch (Req 5.3)
* ═════════════════════════════════════════════════════════════════════════ */
static void test_rdhup_primary_dispatch_gate_opens(void)
{
install_stubs();
cluster_t c;
myconf_t myconf;
make_cluster(&c, &myconf, "ctrl-a");
TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &g_cfg, 0));
ha_do_demote(&c.ha);
c.ha.phase2_done = 1;
c.ha.n_peers = 1;
g_rdhup_state.c = &c;
ha_rdhup_election_hook = rdhup_election_drives_evaluate;
int sv[2];
TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv));
c.ha.primary_ctrl_fd = sv[0];
ha_on_primary_ctrl_fd_rdhup(&c.ha, &c);
close(sv[1]);
/* Drive to CTRL_PRIMARY via ack */
ha_repl_ack_received(&c.ha, &c);
int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT((int)CTRL_PRIMARY, role);
/* Verify dispatch gate allows job operations in CTRL_PRIMARY */
int rc = ha_dispatch_gate(&c.ha, OPCode_jobsub);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"OPCode_jobsub must be allowed by dispatch gate in CTRL_PRIMARY (Req 5.3)");
rc = ha_dispatch_gate(&c.ha, OPCode_jobdispatch);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"OPCode_jobdispatch must be allowed by dispatch gate in CTRL_PRIMARY (Req 5.3)");
clear_stubs();
ha_fini(&c.ha);
printf("test_rdhup_primary_dispatch_gate_opens: PASS (gate accepts job ops)\n");
}
/* ── Suite entry point ──────────────────────────────────────────────────── */
void test_ha_rdhup_integration(void)
{
RUN_TEST(test_rdhup_full_path_reaches_committing);
RUN_TEST(test_rdhup_committing_to_primary_on_ack);
RUN_TEST(test_rdhup_single_node_reaches_primary);
RUN_TEST(test_rdhup_3node_one_ack_promotes);
RUN_TEST(test_rdhup_committing_fires_replication_hooks);
RUN_TEST(test_rdhup_primary_dispatch_gate_opens);
}

197
src/controller/testsuite/test_ha_role.c
View file

@ -88,72 +88,8 @@ static void test_ha_promote_sets_primary_role(void)
printf("test_ha_promote_sets_primary_role: PASS\n");
}
/* ── P1-3 watchdog tests ────────────────────────────────────── */
/* ha_start_watchdog() must be callable after ha_init() */
static void test_watchdog_starts_and_stops(void)
{
ha_state_t ha;
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
};
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &cfg, 0));
TEST_ASSERT_EQUAL_INT(0, ha_start_watchdog(&ha));
/* brief sleep to let the thread wake at least once */
struct timespec ts = { 0, 50 * 1000000L }; /* 50ms */
nanosleep(&ts, NULL);
ha_stop_watchdog(&ha); /* sets stop flag + pthread_join */
ha_fini(&ha);
printf("test_watchdog_starts_and_stops: PASS\n");
}
/* ha_check_heartbeat_nocluster(): when in CTRL_STANDBY with no primary_ep,
* must not promote */
static void test_heartbeat_no_action_when_no_primary(void)
{
ha_state_t ha;
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
};
ha_init(&ha, &cfg, 1 /* not highest priority */);
/* No primary_host set — watchdog must not promote */
ha_check_heartbeat_nocluster(&ha); /* test-only accessor */
TEST_ASSERT_EQUAL_INT(CTRL_STANDBY,
(int)atomic_load_explicit(&ha.role, memory_order_acquire));
ha_fini(&ha);
printf("test_heartbeat_no_action_when_no_primary: PASS\n");
}
/* ha_check_heartbeat_nocluster(): when last_heartbeat is stale (timeout exceeded)
* and primary_host is empty (no known primary), must not crash or change role */
static void test_heartbeat_stale_no_crash_without_primary(void)
{
ha_state_t ha;
ha_config_t cfg = {
.heartbeat_interval_ms = 2000, /* minimum valid */
.heartbeat_timeout_ms = 3000, /* >= interval */
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
};
/* Staleness is injected manually below; timeout value only needs to be valid */
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &cfg, 0));
/* Set last_heartbeat to a time far in the past to simulate timeout */
clock_gettime(CLOCK_MONOTONIC, &ha.last_heartbeat);
ha.last_heartbeat.tv_sec -= 10; /* 10 seconds ago */
ha_check_heartbeat_nocluster(&ha);
/* Should not crash. Role may or may not change — but no segfault. */
ha_fini(&ha);
printf("test_heartbeat_stale_no_crash_without_primary: PASS\n");
}
/* Watchdog / SIGUSR1 tests deleted — watchdog pthread replaced by
* ha_arm_heartbeat_timeout_vtask (delaymgr); failover_requested removed */
/* ── P1-3a tests ─────────────────────────────────────────────── */
@ -206,120 +142,9 @@ static void test_demote_sets_standby_role(void)
printf("test_demote_sets_standby_role: PASS\n");
}
/* SIGUSR1 must set failover_requested via async-signal-safe atomic store */
static void test_sigusr1_sets_failover_requested(void)
{
ha_state_t ha;
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
};
ha_init(&ha, &cfg, 0);
/* test_sigusr1_sets_failover_requested deleted — SIGUSR1 / failover_requested removed */
/* Install real SIGUSR1 handler that uses only the atomic flag */
ha_install_sigusr1(&ha);
/* Send SIGUSR1 to self */
raise(SIGUSR1);
/* Handler must have set failover_requested */
TEST_ASSERT_TRUE(
atomic_load_explicit(&ha.failover_requested, memory_order_acquire) != 0);
/* Restore default SIGUSR1 handler */
signal(SIGUSR1, SIG_DFL);
ha_fini(&ha);
printf("test_sigusr1_sets_failover_requested: PASS\n");
}
/* ── P1-6: am_i_primary() election tests ────────────────────────────── */
/* test: priority-0 with no peers → immediately primary (no probing needed) */
void test_am_i_primary_p0_no_peers(void)
{
cluster_t c;
memset(&c, 0, sizeof(c));
myconf_t myconf;
memset(&myconf, 0, sizeof(myconf));
c.myconf = &myconf;
ha_config_t cfg = {
.heartbeat_interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
.heartbeat_timeout_ms = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
.election_timeout_ms = 500, /* minimum valid value */
.probe_connect_timeout_ms = 100,
.max_promotion_retries = 1,
.election_jitter_max_ms = 0,
.election_priority_step_ms = 0,
};
TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &cfg, 0));
TEST_ASSERT_EQUAL_INT(1, am_i_primary(&c));
ha_fini(&c.ha);
printf("test_am_i_primary_p0_no_peers: PASS\n");
}
/* test: priority-0 with peers → still primary (never probes higher-priority nodes) */
void test_am_i_primary_p0_with_peers(void)
{
cluster_t c;
memset(&c, 0, sizeof(c));
myconf_t myconf;
memset(&myconf, 0, sizeof(myconf));
myconf.peers_cfg.candidates_count = 2;
strncpy(myconf.peers_cfg.candidates[0].hostname, "127.0.0.1", XCE_MAXNAME_LEN - 1);
myconf.peers_cfg.candidates[0].port = 19900;
strncpy(myconf.peers_cfg.candidates[1].hostname, "127.0.0.1", XCE_MAXNAME_LEN - 1);
myconf.peers_cfg.candidates[1].port = 19901;
c.myconf = &myconf;
ha_config_t cfg = {
.heartbeat_interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
.heartbeat_timeout_ms = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
.election_timeout_ms = 500, /* minimum valid value */
.probe_connect_timeout_ms = 100,
.max_promotion_retries = 1,
.election_jitter_max_ms = 0,
.election_priority_step_ms = 0,
};
TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &cfg, 0));
TEST_ASSERT_EQUAL_INT(1, am_i_primary(&c));
ha_fini(&c.ha);
printf("test_am_i_primary_p0_with_peers: PASS\n");
}
/* test: priority-1 with no higher-priority peer alive → deadline expires → primary */
void test_am_i_primary_p1_deadline_expires(void)
{
cluster_t c;
memset(&c, 0, sizeof(c));
myconf_t myconf;
memset(&myconf, 0, sizeof(myconf));
/* Point to ports where nothing is listening */
myconf.peers_cfg.candidates_count = 2;
strncpy(myconf.peers_cfg.candidates[0].hostname, "127.0.0.1", XCE_MAXNAME_LEN - 1);
myconf.peers_cfg.candidates[0].port = 19910; /* nothing listening */
strncpy(myconf.peers_cfg.candidates[1].hostname, "127.0.0.1", XCE_MAXNAME_LEN - 1);
myconf.peers_cfg.candidates[1].port = 19911;
c.myconf = &myconf;
ha_config_t cfg = {
.heartbeat_interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
.heartbeat_timeout_ms = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
.election_timeout_ms = 600, /* expires quickly; > 500ms floor */
.probe_connect_timeout_ms = 200,
.max_promotion_retries = 1,
.election_jitter_max_ms = 0,
.election_priority_step_ms = 0,
};
TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &cfg, 1)); /* priority 1 — must probe priority 0 */
int result = am_i_primary(&c);
TEST_ASSERT_EQUAL_INT(1, result); /* no peer responded → self-declare primary */
ha_fini(&c.ha);
printf("test_am_i_primary_p1_deadline_expires: PASS\n");
}
/* am_i_primary() tests deleted — function removed (election-iomgr-refactor) */
/* ── Task 2.1: ha_state_t new fields zero-initialised by ha_init ──────── */
@ -376,22 +201,12 @@ void test_ha_role(void)
RUN_TEST(test_ha_init_rejects_short_election_timeout);
RUN_TEST(test_ha_init_rejects_timeout_less_than_interval);
RUN_TEST(test_ha_promote_sets_primary_role);
RUN_TEST(test_watchdog_starts_and_stops);
RUN_TEST(test_heartbeat_no_action_when_no_primary);
RUN_TEST(test_heartbeat_stale_no_crash_without_primary);
/* P1-3a */
RUN_TEST(test_promote_increments_term);
RUN_TEST(test_demote_sets_standby_role);
RUN_TEST(test_sigusr1_sets_failover_requested);
/* Watchdog / SIGUSR1 tests deleted */
/* Task 2.1: election-iomgr-refactor new fields */
RUN_TEST(test_ha_init_zeroes_new_fields);
}
/* ── P1-6 suite: am_i_primary() election ─────────────────────────────── */
void test_am_i_primary(void)
{
RUN_TEST(test_am_i_primary_p0_no_peers);
RUN_TEST(test_am_i_primary_p0_with_peers);
RUN_TEST(test_am_i_primary_p1_deadline_expires);
}
/* test_am_i_primary suite deleted — am_i_primary() removed (election-iomgr-refactor) */

22
src/controller/testsuite/test_p1_integration.c
View file

@ -53,27 +53,6 @@ static void test_term_fencing_monotonic(void)
printf("test_term_fencing_monotonic: PASS\n");
}
/* Test 3: Watchdog start+stop cycle — verify clean teardown */
static void test_watchdog_lifecycle_clean(void)
{
ha_state_t ha;
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
};
ha_init(&ha, &cfg, 0);
TEST_ASSERT_EQUAL_INT(0, ha_start_watchdog(&ha));
struct timespec ts = { 0, 10 * 1000000L }; /* 10 ms */
nanosleep(&ts, NULL);
ha_stop_watchdog(&ha);
/* After stop, watchdog_thread must be reset to 0 */
TEST_ASSERT_EQUAL_INT(0, (int)ha.watchdog_thread);
ha_fini(&ha);
printf("test_watchdog_lifecycle_clean: PASS\n");
}
/* Test 4: Redirect guard — primary never redirects; standby with known
* primary redirects; standby with no known primary reports nomaster */
@ -145,7 +124,6 @@ void test_p1_integration(void)
{
RUN_TEST(test_single_node_starts_as_primary);
RUN_TEST(test_term_fencing_monotonic);
RUN_TEST(test_watchdog_lifecycle_clean);
RUN_TEST(test_redirect_guard_integration);
RUN_TEST(test_hooks_registered_guard_integration);
}

25
src/controller/testsuite/test_p3_hardening.c
View file

@ -142,30 +142,6 @@ static void test_p3_promote_demote_cycle_consistent_role(void)
printf("test_p3_promote_demote_cycle_consistent_role: PASS\n");
}
/* Heartbeat timeout with a known primary must set failover_requested
* (replicates the critical path exercised by the watchdog). */
static void test_p3_heartbeat_timeout_sets_failover_flag(void)
{
ha_state_t ha;
ha_config_t cfg = make_p3_ha_cfg();
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &cfg, 1 /* not highest priority */));
/* Seed a known primary so the timeout check fires */
strncpy(ha.primary_host, "dead-primary", XCE_MAXNAME_LEN - 1);
ha.primary_port = 7788;
/* Force an expired heartbeat timestamp */
clock_gettime(CLOCK_MONOTONIC, &ha.last_heartbeat);
ha.last_heartbeat.tv_sec -= 30; /* 30 s ago — far past 10 s timeout */
ha_check_heartbeat_nocluster(&ha);
TEST_ASSERT_TRUE(
atomic_load_explicit(&ha.failover_requested, memory_order_acquire) != 0);
ha_fini(&ha);
printf("test_p3_heartbeat_timeout_sets_failover_flag: PASS\n");
}
/* ================================================================
* P3-2: Redirect state-machine corner cases
@ -478,7 +454,6 @@ void test_p3_hardening(void)
/* P3-1: Failover state machine */
printf("-- P3-1: Failover state machine --\n");
RUN_TEST(test_p3_promote_demote_cycle_consistent_role);
RUN_TEST(test_p3_heartbeat_timeout_sets_failover_flag);
/* P3-2: Redirect corner cases */
printf("-- P3-2: Redirect --\n");