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

refactor(ha): passive standby model — primary connects outbound to all peers

Browse source 
Remove fetch_config_from_primary, ha_find_and_connect_to_primary, and the
tier-1/2/3 discovery path. Standbys never initiate outbound connections to
the primary; the primary owns all inter-controller channels.

Key changes:
- ha.c: add primary heartbeat loop (ha_start_primary_heartbeat) that sends
  OPCode_nop to connected standbys every heartbeat_interval_ms; loop
  self-cancels on demotion via role check before re-arm.
- ha.c: add outbound reconnect in primary_heartbeat_fn — for peers absent
  from standby_endpoints, open fresh outbound TLS connection (on_standby_connected
  sends ctrlhello; standby_ep_ctx_t mirrors vote_ep_ctx_t layout).
- ha.c: remove ha_find_and_connect_to_primary + helpers (ha_raw_tcp_connect,
  ha_tcp_connect_hook); remove election_timeout_fn (superseded by
  raft_election_timeout_fn); remove fetch_config_from_primary call from
  ha_elect_evaluate_and_signal STANDBY branch.
- cluster.c: OPCode_requestVote denial no longer reuses inbound connection;
  late joiners are bootstrapped via primary heartbeat outbound connect.
- config_repl.c: remove connect_ssl_raw, do_one_config_fetch,
  fetch_config_from_primary (~250 LOC).
- myapp.c: remove my_tag > 0 fetch branch; all standbys wait passively.
- ha.h: remove ha_find_and_connect_to_primary declaration + macros +
  ha_tcp_connect_hook seam; add ha_start_primary_heartbeat declaration.
- meson.build: remove test_ha_connect.c (tests deleted function).
- controller_test.c: remove test_ha_connect registration.
- CLAUDE.md, src/controller/CLAUDE.md: document passive standby model,
  inbound-vs-outbound connection ownership, async I/O constraint,
  heartbeat model.
- request.h: add xce_request declaration (removes scattered local externs).
This commit is contained in:
Rongsong Shen 2026-06-01 04:17:05 +00:00
parent 842b9a838e
commit 7b39650229
11 changed files with 716 additions and 919 deletions
Download patch file Download diff file Expand all files Collapse all files

29
CLAUDE.md
View file

@ -64,6 +64,35 @@ Request → Gin Router → Middleware (CORS/RateLimit/IPWhitelist/JWT) → RBAC
- **Pluggable auth**: configfile / LDAP / OAuth2 / PAM (set in `config/auth.yaml`)
- **RBAC**: domain-based role-action-domain model loaded from `config/rbac.yaml`
## Connection Model and Async I/O
### Long-Lived Persistent Connections (CRITICAL)
All XCE daemon connections are **long-lived** — they are established once and reused for the lifetime of the role relationship:
| Connection | Direction | Owner | Purpose |
|---|---|---|---|
| Controller primary → standby | Primary dials out during election | Primary (`ha->standby_endpoints`) | Config push, WAL replication, heartbeat |
| Controller standby ← primary | Inbound on standby listener | Standby (`primary_ctrl_fd`) | Receives ctrlhello, WAL, config, heartbeat |
| Primary → worker | Primary dials out | Primary (`c->workers[i].ep`) | Job dispatch, heartbeat |
| Scheduler → primary | Scheduler dials out | Scheduler | Schedule requests |
| API server → primary | API server dials out | API server backend | Job submit, cluster queries |
Connections are **never** closed and re-opened between RPCs. Closing a connection signals a failure event (RDHUP) which triggers failover or reconnect logic.
### Async Non-Blocking I/O Model (CRITICAL)
All C daemon I/O runs through **iomgr** (epoll-based) + **vtaskmgr** (thread pool):
- **iomgr** — single-threaded epoll loop; dispatches inbound packets as vtasks; manages endpoint lifetime
- **vtaskmgr** — thread-pool executing vtask handlers; handlers must return quickly (no blocking I/O)
- **delaymgr** — timer thread for periodic/delayed tasks; the only place blocking operations (connect, file I/O) are permitted
- `xce_request(ep, &req)` — enqueues a packet to the endpoint's send buffer; **always non-blocking**
- `xce_reply(ep, host, &reply)` — enqueues a reply; **always non-blocking**
**Never block in a vtask handler.** Blocking calls (connect, read, usleep) belong in delaymgr tasks or dedicated threads.
### Primary Heartbeat to Standbys
The primary sends `OPCode_nop` to each standby every `heartbeat_interval_ms` via a self-perpetuating delaymgr one-shot task (`ha_start_primary_heartbeat`). The task re-arms itself only while role is PRIMARY or PRIMARY_COMMITTING — on demotion the loop stops automatically (no cancel API needed). This keeps the standby-side watchdog (`ha_arm_heartbeat_timeout_vtask`) from timing out during idle periods.
## Key Files
- `src/frontend/Makefile` — primary build/test orchestration
- `src/frontend/apiserver/cmd/main.go` — server startup, init sequence

36
src/controller/CLAUDE.md
View file

@ -25,6 +25,42 @@ Links against `libxce` from `src/lib/`.
- Component-specific headers in `include/`
- Shared types defined in `src/include/xce/` (libxce)
## HA Design Constraints
### Inbound vs. Outbound Connection Ownership (CRITICAL)
**Strict rule: inbound connections are for voting only. All persistent control channels are outbound from the primary.**
| Connection type | Direction | Purpose | Who closes |
|---|---|---|---|
| Vote RPC | Standby → Primary (outbound from standby) | `OPCode_requestVote` + reply only | Remote closes after reply |
| Standby control channel | Primary → Standby (outbound from primary) | ctrlhello, config push, WAL replication, heartbeat | `vote_ep_err_handler` / `standby_ep_err_handler_primary` on disconnect |
Rules:
- **Inbound vote connections** (accepted on the listener) are used **exclusively** for the vote exchange. After the reply is sent the primary does NOT reuse the connection for config/replication. The inbound ep is left open until the remote closes it.
- **Primary never uses an inbound connection to push state.** All state push (ctrlhello, config, WAL, heartbeat nop) goes over **outbound connections** stored in `ha->standby_endpoints`.
- During election: candidate opens outgoing vote connections (one per peer). After winning, `raft_cb_become_leader` promotes surviving vote connections to `standby_endpoints`.
- Late joiners: the primary's heartbeat task (`primary_heartbeat_fn`) scans the peers list and **opens a fresh outbound connection** to any peer absent from `standby_endpoints`. On connect, `on_standby_connected` sends ctrlhello and registers the ep.
- `CTRL_STANDBY` must not call `ha_find_and_connect_to_primary` or schedule `fetch_config_from_primary` after an election result — the primary will connect to it.
### Asynchronous, Non-Blocking I/O (CRITICAL)
All inter-controller communication uses iomgr (epoll-based) and vtaskmgr:
- `iomgr` dispatches inbound packets to vtask handlers — never block inside a vtask.
- `vtaskmgr` runs handlers on a thread pool; each handler must return quickly.
- Outgoing sends use `xce_request(ep, &req)` which enqueues to the endpoint's send buffer — non-blocking.
- 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.
### Connection Ownership
- Promoted vote connections (`ha->standby_endpoints`): owned by `vote_ep_err_handler`; freed when the standby disconnects.
- Late-joiner inbound connections: owned by the cluster endpoint array (`c->endpoints`); the `ep_private_t` role identifies `HostRole_controller` for cleanup in `handle_endpoint_error`.
- `ha->primary_ctrl_fd`: set by `handle_ctrlhello` on the standby when the first ctrlhello arrives; used by `ha_on_primary_ctrl_fd_rdhup` to trigger re-election on disconnect.
## Key Files
- `src/controller/` — Controller source files
- `include/` — Controller-specific headers

8
src/controller/cluster.c
View file

@ -322,6 +322,11 @@ static int myreq_handler(vtask_t* task)
reply.data_which = RequestReply_data_requestvotereplymsg;
reply.data.requestvotereplymsg = &vr_rep;
xce_reply(ep, c->myhost, &reply);
/* Inbound vote connections are for voting only. After the reply is
* sent the connection is left open until the remote side closes it.
* The primary will connect outbound to the newcomer from the heartbeat
* task if the peer is missing from standby_endpoints. */
rc = 0;
break;
}
@ -1258,9 +1263,6 @@ int cluster_run(cluster_t* c)
return 0;
}
/* xce_request: defined in request.c, not yet in any public header */
extern int xce_request(endpoint_t *ep, xce_request_t *req);
/* ── Task 4.2: ha_send_ctrlhello_to_all_standbys ───────────────────────────
*
* Send OP_hello CtrlHelloRequest to every connected standby endpoint.

436
src/controller/config_repl.c
View file

@ -37,9 +37,6 @@
#include "xce/cluster_def.h"
#include "wal_types.h" /* wal_entry_hdr_t, WAL_* constants */
/* xce_request: defined in request.c, not yet in any public header */
extern int xce_request(endpoint_t *ep, xce_request_t *req);
/* SSL context for xce_probe_config_tag.
* Stored as void* to avoid SSL_CTX type at this point in the translation unit. */
static void *s_probe_ssl_ctx = NULL;
@ -678,320 +675,6 @@ int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req)
return rc2;
}
/* ── fetch_config_from_primary helpers ─────────────────────── */
/*
* connect_ssl_raw: open a non-blocking TCP connection to host:port, then
* perform a TLS handshake using ssl_ctx.
*
* The poll loop checks ha_is_primary(ha) between iterations: if this node
* is promoted to primary while waiting, the function returns -ECANCELED so
* the caller can switch to the primary path immediately.
*
* On success: *out_ssl and *out_sock are set; caller owns them and must call
* SSL_shutdown(ssl) + SSL_free(ssl) + close(sock).
* On failure: nothing is left open; returns negative errno.
*/
static int connect_ssl_raw(SSL_CTX *ssl_ctx, const char *host, uint32_t port,
int poll_timeout_ms, const ha_state_t *ha,
SSL **out_ssl, int *out_sock)
{
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)port);
if (getaddrinfo(host, port_str, &hints, &res) != 0)
return -ENONET;
int sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (sock < 0) {
int e = errno;
freeaddrinfo(res);
return -e;
}
/* 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 ha_is_primary between iterations */
while (!ha_is_primary(ha)) {
struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP };
int pr = poll(&pfd, 1, poll_timeout_ms);
if (pr < 0 && errno == EINTR)
continue;
if (pr == 0) {
close(sock);
return -ETIMEDOUT;
}
if (pfd.revents & (POLLERR | POLLHUP)) {
close(sock);
return -ECONNREFUSED;
}
if (pfd.revents & POLLOUT)
break;
}
if (ha_is_primary(ha)) {
close(sock);
return -ECANCELED;
}
/* Verify connect succeeded */
{
int err = 0;
socklen_t el = sizeof(err);
getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
if (err != 0) {
close(sock);
return -err;
}
}
/* Restore blocking for SSL handshake */
fcntl(sock, F_SETFL, flags & ~O_NONBLOCK);
SSL *ssl = SSL_new(ssl_ctx);
if (!ssl) {
close(sock);
return -ENOMEM;
}
SSL_set_fd(ssl, sock);
int sr;
do {
sr = SSL_connect(ssl);
if (sr <= 0) {
int e = SSL_get_error(ssl, sr);
if (e != SSL_ERROR_WANT_READ && e != SSL_ERROR_WANT_WRITE)
break;
}
} while (sr <= 0 && !ha_is_primary(ha));
if (sr <= 0 || ha_is_primary(ha)) {
SSL_free(ssl);
close(sock);
return -ECONNRESET;
}
*out_ssl = ssl;
*out_sock = sock;
return 0;
}
/*
* do_one_config_fetch: attempt one network fetch of the ClusterConfig from
* the primary controller over a standalone SSL socket.
*
* Steps:
* 1. Open non-blocking TCP + SSL handshake via connect_ssl_raw()
* 2. Send OP_replConfig request with our current version tag
* 3. Decode the reply
* 4. If up_to_date: nothing to save — return 0
* 5. Verify SHA-256 integrity of received bytes
* 6. Save to disk via conf_save_config()
* 7. Reload cluster_config via load_primary_config()
* 8. Log success and clean up
*
* Returns 0 on success (config up-to-date or freshly saved+loaded).
* Returns -ECANCELED if promoted to primary during the attempt (caller should
* switch to primary path). Returns other negative errno on transient failure
* (caller will retry with backoff).
*/
static int do_one_config_fetch(cluster_t *c, const char *host, uint32_t port,
uint64_t our_tag)
{
SSL *ssl = NULL;
int sock = -1;
/* 1. Connect + SSL handshake — use client_ssl_ctx (TLS_client_method) for
* outgoing connections; ssl_ctx is the server context (TLS_server_method). */
SSL_CTX *out_ctx = c->ha.client_ssl_ctx
? (SSL_CTX *)c->ha.client_ssl_ctx
: (SSL_CTX *)c->ha.ssl_ctx;
int rc = connect_ssl_raw(
out_ctx, host, port,
(int)c->ha.config.probe_connect_timeout_ms,
&c->ha, &ssl, &sock);
if (rc != 0)
return rc;
/* 2. Send OP_replConfig request */
rc = xce_replconfig_req_ssl(ssl, our_tag);
if (rc != 0) {
SSL_shutdown(ssl);
SSL_free(ssl);
close(sock);
return rc;
}
/* 3. Decode reply */
int up_to_date = 0;
uint64_t new_tag = 0;
uint8_t *sha256 = NULL;
uint8_t *config_bytes = NULL;
size_t config_len = 0;
rc = xce_replconfig_reply_ssl_decode(ssl, &up_to_date, &new_tag,
&sha256, &config_bytes, &config_len);
if (rc != 0) {
SSL_shutdown(ssl);
SSL_free(ssl);
close(sock);
return rc;
}
/* 4. Up-to-date fast path: our tag matches primary — nothing to save */
if (up_to_date) {
SSL_shutdown(ssl);
SSL_free(ssl);
close(sock);
return 0;
}
/* 5. Verify SHA-256 integrity */
uint8_t computed[32];
SHA256(config_bytes, config_len, computed);
if (memcmp(computed, sha256, 32) != 0) {
ERROR("HA: config fetch SHA-256 mismatch from primary — retrying");
free(sha256);
free(config_bytes);
SSL_shutdown(ssl);
SSL_free(ssl);
close(sock);
return -EBADMSG;
}
/* 6. Save to disk atomically */
rc = conf_save_config(c->myconf, config_bytes, config_len, sha256);
free(sha256);
free(config_bytes);
sha256 = NULL;
config_bytes = NULL;
if (rc != 0) {
SSL_shutdown(ssl);
SSL_free(ssl);
close(sock);
return rc;
}
/* 7. Reload cluster_config from the newly saved file */
rc = load_primary_config(c);
if (rc != 0) {
SSL_shutdown(ssl);
SSL_free(ssl);
close(sock);
return rc;
}
/* 8. Log success */
INFO("HA: config received from primary (tag=%016" PRIx64 ")",
c->myconf->config_version_tag);
/* Clean up SSL and socket */
SSL_shutdown(ssl);
SSL_free(ssl);
close(sock);
return 0;
}
/* ── fetch_config_from_primary ──────────────────── */
int fetch_config_from_primary(cluster_t *c)
{
const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
/* Build conf dir path: <workdir>/<name> */
char conf_dir[4096];
snprintf(conf_dir, sizeof(conf_dir), "%s/%s",
c->myconf->workdir, c->myconf->name);
uint64_t saved_tag = conf_get_saved_tag(conf_dir);
if (pcfg->candidates_count == 0) {
/* Single-node mode — no primary to contact; load from saved file */
goto use_saved;
}
{
/* Primary is candidates[0] (highest priority = lowest index) */
const char *primary_host = pcfg->candidates[0].hostname;
uint32_t primary_port = pcfg->candidates[0].port;
INFO("HA: requesting config from primary %s:%u (my_tag=%016" PRIx64 ")",
primary_host, primary_port, saved_tag);
uint32_t delay_ms = 500;
const uint32_t max_delay_ms = 5000;
const int max_attempts = 60; /* up to 5min: primary may still be initializing */
for (int attempt = 0; attempt < max_attempts; attempt++) {
/* If promoted during retry, switch to primary path */
if (ha_is_primary(&c->ha)) {
INFO("HA: promoted to primary during config fetch"
" — switching to primary path");
return load_primary_config(c);
}
int rc = do_one_config_fetch(c, primary_host, primary_port,
saved_tag);
if (rc == 0) {
INFO("HA: config received from primary"
" (tag=%016" PRIx64 ")",
c->myconf->config_version_tag);
/* Config loaded — complete standby bootstrap:
* 1. cluster_init_phase2 to set up data structures with config
* 2. ha_find_and_connect_to_primary to establish the persistent
* control channel needed for RDHUP failover detection */
#ifndef XCE_TESTING
{
extern int cluster_init_phase2(cluster_t *c);
int p2rc = cluster_init_phase2(c);
if (p2rc == 0) {
ha_find_and_connect_to_primary(&c->ha, &c->ha.config,
&c->myconf->peers_cfg, c);
}
}
#endif
return 0;
}
if (attempt < max_attempts - 1) {
WARN("HA: config fetch attempt %d/%d failed (rc=%d)"
" — retrying in %ums",
attempt + 1, max_attempts, rc, delay_ms);
usleep((useconds_t)delay_ms * 1000);
delay_ms = delay_ms * 2 < max_delay_ms
? delay_ms * 2 : max_delay_ms;
}
}
}
use_saved:
{
char fname[256] = {0};
int rc = conf_read_current_filename(conf_dir, fname, sizeof(fname));
if (rc != 0) {
ERROR("Standby (priority=%d) could not fetch config from primary"
" and has no saved config at '%s/conf/current'"
" — start the primary first or provision config manually",
c->ha.my_priority, conf_dir);
return -ENOENT;
}
WARN("HA: using saved config (tag=%016" PRIx64 ")"
" — config may be stale", saved_tag);
/* load_primary_config resolves conf/current → versioned file,
* decodes Cap'n Proto, computes SHA-256, and checks staleness.
* It is reused here since the saved-config load path is identical. */
return load_primary_config(c);
}
}
/* ── send_upgrade_error ─────────────────────────── */
static void send_upgrade_error(endpoint_t *ep, xce_request_t *req,
@ -1399,43 +1082,58 @@ int xce_probe_config_tag(const char *host, uint16_t port, uint64_t *tag_out)
int sock = -1;
int flags_saved = 0;
/* ── Step 1: SSL connect via connect_ssl_raw (same path as do_one_config_fetch) ── */
/* ── Step 1: TCP connect (+ optional SSL handshake) ── */
{
/* s_ssl_ctx may be set externally; if NULL (unit test), fall back to plain TCP. */
SSL_CTX *ssl_ctx = s_probe_ssl_ctx;
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)port);
if (getaddrinfo(host, port_str, &hints, &res) != 0)
return -EHOSTUNREACH;
sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (sock < 0) { int e = errno; freeaddrinfo(res); return -e; }
flags_saved = fcntl(sock, F_GETFL, 0);
fcntl(sock, F_SETFL, flags_saved | O_NONBLOCK);
connect(sock, res->ai_addr, res->ai_addrlen);
freeaddrinfo(res);
struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP };
int pr;
do { pr = poll(&pfd, 1, 500); } while (pr < 0 && errno == EINTR);
if (pr == 0) { close(sock); return -ETIMEDOUT; }
if (pr < 0 || (pfd.revents & (POLLERR | POLLHUP))) { close(sock); return -ECONNREFUSED; }
int err = 0; socklen_t el = sizeof(err);
getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
if (err) { close(sock); return -err; }
fcntl(sock, F_SETFL, flags_saved & ~O_NONBLOCK);
/* Optional SSL handshake — only when s_probe_ssl_ctx is set */
SSL_CTX *ssl_ctx = (SSL_CTX *)s_probe_ssl_ctx;
if (ssl_ctx) {
int rc = connect_ssl_raw(ssl_ctx, host, port, 500, NULL, &ssl, &sock);
if (rc != 0)
return rc;
} else {
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)port);
if (getaddrinfo(host, port_str, &hints, &res) != 0)
return -EHOSTUNREACH;
sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (sock < 0) { int e = errno; freeaddrinfo(res); return -e; }
flags_saved = fcntl(sock, F_GETFL, 0);
fcntl(sock, F_SETFL, flags_saved | O_NONBLOCK);
connect(sock, res->ai_addr, res->ai_addrlen);
freeaddrinfo(res);
struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP };
int pr;
do { pr = poll(&pfd, 1, 500); } while (pr < 0 && errno == EINTR);
if (pr == 0) { close(sock); return -ETIMEDOUT; }
if (pr < 0 || (pfd.revents & (POLLERR | POLLHUP))) { close(sock); return -ECONNREFUSED; }
int err = 0; socklen_t el = sizeof(err);
getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
if (err) { close(sock); return -err; }
fcntl(sock, F_SETFL, flags_saved & ~O_NONBLOCK);
ssl = SSL_new(ssl_ctx);
if (!ssl) { close(sock); return -ENOMEM; }
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);
if (sr <= 0) {
SSL_free(ssl);
ssl = NULL;
close(sock);
return -ECONNREFUSED;
}
}
}
@ -1472,11 +1170,18 @@ int xce_probe_config_tag(const char *host, uint16_t port, uint64_t *tag_out)
/* Wire framing: 8-byte size prefix then capnp payload (matches ha.c pattern) */
uint64_t hdr = (uint64_t)blen;
ssize_t nsent;
nsent = (ssize_t)write(sock, &hdr, sizeof(hdr));
if (nsent == (ssize_t)sizeof(hdr))
nsent = (ssize_t)write(sock, buf, (size_t)blen);
if (ssl) {
nsent = (ssize_t)SSL_write(ssl, &hdr, (int)sizeof(hdr));
if (nsent == (ssize_t)sizeof(hdr))
nsent = (ssize_t)SSL_write(ssl, buf, (int)blen);
} else {
nsent = (ssize_t)write(sock, &hdr, sizeof(hdr));
if (nsent == (ssize_t)sizeof(hdr))
nsent = (ssize_t)write(sock, buf, (size_t)blen);
}
free(buf);
if (nsent != (ssize_t)blen) {
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
close(sock);
return -EIO;
}
@ -1489,8 +1194,10 @@ int xce_probe_config_tag(const char *host, uint16_t port, uint64_t *tag_out)
/* Read 8-byte size header */
while (off < sizeof(hdr)) {
int n = (int)read(sock, (char *)&hdr + off, sizeof(hdr) - off);
int n = ssl ? SSL_read(ssl, (char *)&hdr + off, (int)(sizeof(hdr) - off))
: (int)read(sock, (char *)&hdr + off, sizeof(hdr) - off);
if (n <= 0) {
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
close(sock);
return -EIO;
}
@ -1498,21 +1205,25 @@ int xce_probe_config_tag(const char *host, uint16_t port, uint64_t *tag_out)
}
if (hdr == 0 || hdr > 65536) { /* sanity cap — reply is tiny */
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
close(sock);
return -EMSGSIZE;
}
uint8_t *buf = (uint8_t *)malloc((size_t)hdr);
if (!buf) {
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
close(sock);
return -ENOMEM;
}
off = 0;
while (off < (size_t)hdr) {
int n = (int)read(sock, buf + off, (size_t)hdr - off);
int n = ssl ? SSL_read(ssl, buf + off, (int)((size_t)hdr - off))
: (int)read(sock, buf + off, (size_t)hdr - off);
if (n <= 0) {
free(buf);
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
close(sock);
return -EIO;
}
@ -1536,6 +1247,7 @@ int xce_probe_config_tag(const char *host, uint16_t port, uint64_t *tag_out)
*tag_out = rep_s.configTag;
}
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
close(sock);
INFO("HA: probe config tag from %s:%u tag=%016" PRIx64,
host, (unsigned)port, *tag_out);
@ -1590,6 +1302,18 @@ int handle_ctrlhello(cluster_t *c, endpoint_t *ep, xce_request_t *req)
uint64_t standby_applied_lsn = hello->applied_lsn;
uint64_t standby_config_tag = hello->config_tag;
/* Req 3.1: on the standby side, the first ctrlhello from the winner
* designates this inbound connection as the primary control channel.
* Guard with primary_ctrl_fd == -1 for idempotency. */
if (c && ep &&
(ctrl_role_t)atomic_load_explicit(&c->ha.role,
memory_order_acquire) == CTRL_STANDBY &&
c->ha.primary_ctrl_fd == -1) {
c->ha.primary_ctrl_fd = ep->fd;
ha_on_heartbeat_received(&c->ha);
INFO("HA: primary control channel established (fd=%d)", ep->fd);
}
/* 2. Register standby in replication set and mark as catching up */
int rc = repl_add_standby(&c->persist.repl, ep,
hello->name ? hello->name : "unknown", 0);

792
src/controller/ha.c
View file

@ -180,6 +180,9 @@ static int raft_cb_send_vote_request(void *ctx, int peer_idx, uint64_t term)
return 1;
}
/* Track endpoint for promotion/close after election result */
ectx->votes[votes_idx].ep = ep;
return 0;
}
@ -381,6 +384,24 @@ static void raft_cb_become_leader(void *ctx, uint64_t term)
/* Reset ack counter for this committing round */
atomic_store(&ha->repl_ack_count, 0);
/* Promote surviving vote endpoints to standby replication channels.
* Iterate ha->elect_ctx->votes[i].ep and add non-NULL entries to
* ha->standby_endpoints so push_config and ctrlhello reach them. */
{
ha_elect_ctx_t *ectx = (ha_elect_ctx_t *)atomic_load_explicit(
&ha->elect_ctx, memory_order_acquire);
if (ectx && ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
for (int i = 0; i < ectx->n_peers; i++) {
if (ectx->votes[i].ep != NULL) {
zarray_store(ha->standby_endpoints, ectx->votes[i].ep);
INFO("HA: promoted vote ep[%d] → standby channel", i);
}
}
pthread_mutex_unlock(&ha->lock);
}
}
if (c) {
/* Push config to standbys with lower config_tag */
#ifdef XCE_TESTING
@ -447,19 +468,31 @@ static void raft_cb_become_follower(void *ctx, uint64_t term, int leader_idx)
ha->phase2_done = 1;
}
/* Close all outgoing vote endpoints (loser path).
* The primary will push config and ctrlhello via promoted connections;
* the standby waits passively on its listener for the inbound connection. */
{
ha_elect_ctx_t *ectx = (ha_elect_ctx_t *)atomic_load_explicit(
&ha->elect_ctx, memory_order_acquire);
if (ectx) {
for (int i = 0; i < ectx->n_peers; i++) {
if (ectx->votes[i].ep != NULL) {
INFO("HA: closing vote ep[%d] (loser path)", i);
endpoint_fini(ectx->votes[i].ep);
ectx->votes[i].ep = NULL;
}
}
}
}
atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY,
memory_order_release);
INFO("HA FSM (raft): → CTRL_STANDBY (term=%" PRIu64 ", first_boot=%d)",
term, is_first_boot);
/* Trigger async config fetch from primary so standby can load config */
#ifndef XCE_TESTING
if (c != NULL && is_first_boot) {
extern int fetch_config_from_primary(cluster_t *c);
delay_task_ms(&c->delaymgr, 100, 1,
(delay_task_entry_fn_t)fetch_config_from_primary, c);
}
#endif
/* Arm heartbeat watchdog — fires ha_on_primary_ctrl_fd_rdhup if primary
* sends no ctrlhello within election_timeout_ms. */
ha_arm_heartbeat_timeout_vtask(ha, c);
}
/*
@ -1142,6 +1175,11 @@ int ha_promote_to_primary(cluster_t *c)
(hook_fn_t)persist_worker_hook_fun, &c->persist);
c->ha.hooks_registered = 1;
}
/* Arm periodic heartbeat to standbys — keeps their watchdog alive when
* the cluster is idle (no WAL entries or config pushes). */
ha_start_primary_heartbeat(c);
return 0;
}
@ -1495,6 +1533,7 @@ int (*ha_send_ctrlhello_hook)(void *c) = NULL;
* (config_repl.c's push_config_to_standbys needs a real cluster). Req 7.1.
*/
int (*ha_push_config_hook)(void *c) = NULL;
#endif /* XCE_TESTING */
/* ── Election Evaluation (Req 2.1, 2.7, 2.8) ───────────────────────────────
@ -1697,17 +1736,8 @@ apply_role:
INFO("HA FSM: → CTRL_STANDBY (term=%" PRIu64 ", first_boot=%d)",
atomic_load_explicit(&ha->term, memory_order_acquire),
is_first_boot);
/* Trigger async config fetch from primary so standby can load
* cluster_config, establish persistent control channel, and
* participate in RDHUP failover detection. */
#ifndef XCE_TESTING
if (c != NULL) {
extern int fetch_config_from_primary(cluster_t *c);
delay_task_ms(&c->delaymgr, 100, 1,
(delay_task_entry_fn_t)fetch_config_from_primary, c);
}
#endif
/* Primary connects outbound and pushes config via ctrlhello.
* Standby is fully passive here — no fetch_config_from_primary. */
}
}
@ -1796,16 +1826,15 @@ void handle_request_vote_reply(ha_state_t *ha,
/* vote_ep_ctx_t: typedef moved to Task 6.2 forward-declaration block above */
/* xce_request: defined in request.c, not yet in any public header */
extern int xce_request(endpoint_t *ep, xce_request_t *req);
/*
* vote_reply_msg_handler: forward declaration.
* Implemented in task 2.3. Registered on the endpoint by on_vote_connected
* so that the next iomgr EPOLLIN delivers the vote reply to this handler.
* Signature matches msg_handler_t = int (*)(void*) from netio.h.
* peer_ctrl_msg_handler / vote_reply_decode_and_dispatch /
* repl_ack_decode_and_dispatch: forward declarations.
* peer_ctrl_msg_handler is the single iomgr handler set in on_vote_connected
* and never reassigned — it routes by ha->role at dispatch time.
*/
static int vote_reply_msg_handler(void *ep);
static int peer_ctrl_msg_handler(void *ep_raw);
static int vote_reply_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx);
static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx);
#ifdef XCE_TESTING
/*
@ -1865,8 +1894,9 @@ static void on_vote_connected(endpoint_t *ep)
xce_request(ep, &req);
#endif
/* Switch to reply-reading mode: next inbound packet → vote_reply_msg_handler */
ep->msg_handler = vote_reply_msg_handler;
/* Set state-aware dispatcher — routes by ha->role at message-arrival time.
* Never reassigned: ELECTING → vote reply; COMMITTING/PRIMARY → repl ack. */
ep->msg_handler = peer_ctrl_msg_handler;
}
/*
@ -1910,51 +1940,32 @@ static void vote_reply_dispatch(ha_state_t *ha, ha_elect_ctx_t *ctx, int idx,
}
/*
* vote_reply_msg_handler: iomgr message handler for incoming RequestVoteReply.
*
* Called by iomgr (via INVOKE_MSG_HANDLER) when a full packet arrives on an
* outgoing vote endpoint after on_vote_connected fired.
*
* Steps:
* 1. Cast ep to endpoint_t*; get packet from ep->fsm_in.pkt (owned by us).
* 2. Raw capnp decode: init capn_ctx, read outer Request wrapper.
* 3. read_RequestReply() into rep_s (capnp pointer — still valid).
* 4. read_RequestVoteReply() into vr_raw (scalars) BEFORE capn_free.
* 5. capn_free() — capnp segment released; vr_raw still valid (scalars only).
* 6. free_packet() + free(pkt) — packet memory released.
* 7. Term fencing: if vr_raw.term > ha->term, update ha->term atomically.
* 8. Call vote_reply_dispatch() with vr_raw scalar fields.
*
* On any decode error: treat peer as denied (grant=0, tag=0, lsn=0).
*
* ASAN safety: all capnp fields read before capn_free() — no heap-use-after-free.
* Requirements: 3.1 (iomgr packet), 3.2 (no capnp UAF), 3.3 (extract fields),
* 3.4 (term fencing).
* vote_reply_decode_and_dispatch: decode a RequestVoteReply packet and dispatch
* to vote_reply_dispatch. Factored from vote_reply_msg_handler so that
* peer_ctrl_msg_handler can invoke the same logic without code duplication.
* Packet ownership: caller passes ep->fsm_in.pkt; this function frees it.
*/
static int vote_reply_msg_handler(void *ep_raw)
static int vote_reply_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
{
endpoint_t *ep = (endpoint_t *)ep_raw;
vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data;
ha_state_t *ha = vctx->ha;
ha_elect_ctx_t *ctx = vctx->ctx;
int idx = vctx->peer_idx;
io_packet_t *pkt = ep->fsm_in.pkt; /* ownership transferred by INVOKE_MSG_HANDLER */
int ok = 0;
ha_state_t *ha = vctx->ha;
ha_elect_ctx_t *ctx = vctx->ctx;
int idx = vctx->peer_idx;
io_packet_t *pkt = ep->fsm_in.pkt;
int ok = 0;
/* Decoded reply fields — populated on success path */
uint64_t vr_term = 0;
int vr_granted = 0;
uint64_t vr_tag = 0;
uint64_t vr_lsn = 0;
uint64_t vr_term = 0;
int vr_granted = 0;
uint64_t vr_tag = 0;
uint64_t vr_lsn = 0;
if (!pkt || !pkt->data || pkt->sz == 0)
goto done;
{
struct capn capn_ctx;
struct Request outer_rq;
Request_ptr outer_rp;
struct RequestReply rep_s;
struct capn capn_ctx;
struct Request outer_rq;
Request_ptr outer_rp;
struct RequestReply rep_s;
struct RequestVoteReply vr_raw;
int capn_ok;
@ -1979,13 +1990,10 @@ static int vote_reply_msg_handler(void *ep_raw)
goto done;
}
/* CRITICAL: copy all scalars BEFORE capn_free to avoid UAF.
* struct RequestVoteReply fields: term, voteGranted, peerConfigTag,
* peerLastLogLsn — all scalars; safe to copy before freeing segment. */
memset(&vr_raw, 0, sizeof(vr_raw));
read_RequestVoteReply(&vr_raw, rep_s.data.requestvotereplymsg);
capn_free(&capn_ctx); /* capnp segment released; vr_raw.* still valid */
capn_free(&capn_ctx);
vr_term = (uint64_t)vr_raw.term;
vr_granted = vr_raw.voteGranted ? 1 : 0;
@ -1995,14 +2003,9 @@ static int vote_reply_msg_handler(void *ep_raw)
}
done:
/* Release packet — INVOKE_MSG_HANDLER will null ep->fsm_in.pkt after return */
if (pkt) { free_packet(pkt); free(pkt); }
if (ok) {
/* Term fencing: if peer's term > ours, atomically advance ha->term.
* Must be done here (before vote_reply_dispatch) so it applies on both
* the production path (raft_on_vote_reply) and the XCE_TESTING hook path
* where raft_cb_become_follower is bypassed. */
uint64_t cur = atomic_load_explicit(&ha->term, memory_order_acquire);
while (vr_term > cur) {
if (atomic_compare_exchange_weak_explicit(&ha->term, &cur, vr_term,
@ -2012,7 +2015,6 @@ done:
}
vote_reply_dispatch(ha, ctx, idx, vr_term, vr_granted, vr_tag, vr_lsn);
} else {
/* Decode error: treat peer as denied */
vote_reply_dispatch(ha, ctx, idx,
atomic_load_explicit(&ha->term, memory_order_acquire),
0, 0, 0);
@ -2020,11 +2022,87 @@ done:
return 0;
}
/*
* repl_ack_decode_and_dispatch: decode an inbound packet on a promoted vote
* endpoint while role == CTRL_PRIMARY_COMMITTING or CTRL_PRIMARY.
* Expects OPCode_replCatchUpDone; calls ha_repl_ack_received on success.
* Drops unexpected opcodes with a WARN log.
*/
static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
{
ha_state_t *ha = vctx->ha;
cluster_t *c = vctx->c;
io_packet_t *pkt = ep->fsm_in.pkt;
if (!pkt || !pkt->data || pkt->sz == 0) {
if (pkt) { free_packet(pkt); free(pkt); }
return 0;
}
struct capn capn_ctx;
struct Request outer_rq;
Request_ptr outer_rp;
int capn_ok = capn_init_mem(&capn_ctx, (const uint8_t *)pkt->data,
(size_t)pkt->sz, 0);
free_packet(pkt);
free(pkt);
ep->fsm_in.pkt = NULL;
if (capn_ok != 0)
return 0;
outer_rp.p = capn_getp(capn_root(&capn_ctx), 0, 1);
read_Request(&outer_rq, outer_rp);
int opcode = (int)outer_rq.opcode;
capn_free(&capn_ctx);
if (opcode == OPCode_replCatchUpDone) {
ha_repl_ack_received(ha, c);
} else {
WARN("repl_ack_decode_and_dispatch: unexpected opcode %d on promoted ep",
opcode);
}
return 0;
}
/*
* peer_ctrl_msg_handler: state-aware iomgr handler set once in on_vote_connected.
* Routes by ha->role at dispatch time — no explicit handler reassignment needed.
* CTRL_ELECTING → vote_reply_decode_and_dispatch
* CTRL_PRIMARY_COMMITTING /
* CTRL_PRIMARY → repl_ack_decode_and_dispatch
* any other role → drop packet (WARN)
*/
static int peer_ctrl_msg_handler(void *ep_raw)
{
endpoint_t *ep = (endpoint_t *)ep_raw;
vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data;
ha_state_t *ha = vctx->ha;
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
switch (role) {
case CTRL_ELECTING:
return vote_reply_decode_and_dispatch(ep, vctx);
case CTRL_PRIMARY_COMMITTING:
case CTRL_PRIMARY:
return repl_ack_decode_and_dispatch(ep, vctx);
default: {
io_packet_t *pkt = ep->fsm_in.pkt;
if (pkt) { free_packet(pkt); free(pkt); ep->fsm_in.pkt = NULL; }
WARN("peer_ctrl_msg_handler: unexpected role %d, packet dropped", (int)role);
return 0;
}
}
}
#ifdef XCE_TESTING
/*
* vote_reply_build_pkt_for_test: test helper that encodes a RequestVoteReply
* into a newly allocated io_packet_t that vote_reply_msg_handler can decode.
* into a newly allocated io_packet_t that vote_reply_msg_handler_for_test can decode.
*
* Wire format mirrors the real peer reply:
* Request{opcode=reply, body.replymsg=RequestReply{
@ -2096,15 +2174,23 @@ io_packet_t *vote_reply_build_pkt_for_test(uint64_t term, int vote_granted,
}
/*
* vote_reply_msg_handler_for_test: expose vote_reply_msg_handler under
* XCE_TESTING so tests can call it directly (it is otherwise static).
* Sets ep->fsm_in.pkt = pkt then calls vote_reply_msg_handler(ep).
* vote_reply_msg_handler_for_test: test entry point for vote-reply decode path.
* Temporarily forces role to CTRL_ELECTING so peer_ctrl_msg_handler routes to
* vote_reply_decode_and_dispatch, then restores the original role.
* Caller owns ep->data (vote_ep_ctx_t); pkt ownership is transferred.
*/
int vote_reply_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt)
{
vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data;
ha_state_t *ha = vctx ? vctx->ha : NULL;
int32_t saved_role = ha ? atomic_load_explicit(&ha->role, memory_order_acquire) : 0;
if (ha)
atomic_store_explicit(&ha->role, (int32_t)CTRL_ELECTING, memory_order_release);
ep->fsm_in.pkt = pkt;
return vote_reply_msg_handler(ep);
int rc = peer_ctrl_msg_handler(ep);
if (ha)
atomic_store_explicit(&ha->role, saved_role, memory_order_release);
return rc;
}
/*
@ -2181,28 +2267,85 @@ int committing_timeout_fn_for_test(ha_state_t *ha, cluster_t *c,
ha_promote_to_primary(c);
return 0;
}
#endif /* XCE_TESTING */
/* ── Task 2.4: election_timeout_fn — delaymgr callback ─────────────────────
*
* Fires ha_elect_evaluate_and_signal after election_timeout_ms if not cancelled.
* The cancelled flag is set atomically when quorum is already reached, making
* this callback a no-op. Owns the election_timeout_ctx_t — frees it on exit.
/*
* repl_catchup_done_pkt_for_test: build a minimal io_packet_t encoding
* Request{opcode=replCatchUpDone, body_which=nodata}.
* Used by peer_ctrl_msg_handler tests to verify the COMMITTING dispatch path.
*/
static int election_timeout_fn(void *arg)
io_packet_t *repl_catchup_done_pkt_for_test(void)
{
election_timeout_ctx_t *t = (election_timeout_ctx_t *)arg;
if (atomic_load(&t->cancelled)) {
free(t);
return 0;
}
pthread_mutex_lock(&t->ctx->lock);
ha_elect_evaluate_and_signal(t->ha, t->ctx);
pthread_mutex_unlock(&t->ctx->lock);
free(t);
return 0;
struct capn cs;
struct capn_segment *seg;
struct Request rq_capn;
Request_ptr rq_ptr;
io_packet_t *pkt;
int rc;
capn_init_malloc(&cs);
seg = capn_root(&cs).seg;
memset(&rq_capn, 0, sizeof(rq_capn));
rq_capn.opcode = OPCode_replCatchUpDone;
rq_capn.seq = 1;
rq_capn.body_which = Request_body_nomsg;
rq_ptr = new_Request(seg);
write_Request(&rq_capn, rq_ptr);
capn_setp(capn_root(&cs), 0, rq_ptr.p);
pkt = calloc(1, sizeof(*pkt));
if (!pkt) { capn_free(&cs); return NULL; }
pkt->sz = capn_size(&cs);
rc = alloc_packet(pkt);
if (rc < 0) { capn_free(&cs); free(pkt); return NULL; }
capn_write_mem(&cs, (uint8_t *)pkt->data, (size_t)pkt->sz, 0);
capn_free(&cs);
return pkt;
}
/*
* vote_ep_err_handler_for_test: call the static vote_ep_err_handler on ep.
* Allows tests to invoke the err_handler without going through iomgr/endpoint machinery.
*/
int vote_ep_err_handler_for_test(endpoint_t *ep)
{
return vote_ep_err_handler((void *)ep);
}
/*
* peer_ctrl_msg_handler_for_test: expose peer_ctrl_msg_handler for tests.
* Sets ep->fsm_in.pkt then invokes the handler. Packet ownership transferred.
*/
int peer_ctrl_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt)
{
ep->fsm_in.pkt = pkt;
return peer_ctrl_msg_handler(ep);
}
/*
* raft_cb_become_leader_for_test: drive the leader promotion path without
* a live cluster. Passes ha as ctx; sets ha->raft_c = NULL to suppress
* production-only calls (phase2, push_config, ctrlhello, committing timer).
*/
void raft_cb_become_leader_for_test(ha_state_t *ha, uint64_t term)
{
ha->raft_c = NULL;
raft_cb_become_leader((void *)ha, term);
}
/*
* raft_cb_become_follower_for_test: drive the follower path without a live
* cluster. Passes ha as ctx; sets ha->raft_c = NULL to suppress production
* calls (phase2, fetch_config); ha_arm_heartbeat_timeout_vtask is a no-op
* when c=NULL so no watchdog thread is spawned.
*/
void raft_cb_become_follower_for_test(ha_state_t *ha, uint64_t term)
{
ha->raft_c = NULL;
raft_cb_become_follower((void *)ha, term, -1);
}
#endif /* XCE_TESTING */
/* ── Task 4.1: committing_timeout_fn — delaymgr callback ───────────────────
*
* Fires ha_promote_to_primary (degraded mode) if the majority ack threshold
@ -2260,15 +2403,32 @@ static int vote_ep_err_handler(void *ep_raw)
vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data;
if (!vctx) return 0;
handle_request_vote_reply(vctx->ha, vctx->ctx, vctx->peer_idx,
vctx->my_term,
0, /* vote_granted = false */
0, 0);
/* Fix 5 (HIGH): do NOT free vctx here.
* ep->data (vctx) is also referenced by vote_reply_msg_handler via ep->data.
* The endpoint lifetime is managed by iomgr; vctx is freed in
* ha_election_cleanup() (task 5.1) when all per-peer contexts are released.
* Freeing here would cause a use-after-free in vote_reply_msg_handler. */
ha_state_t *ha = vctx->ha;
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
if (role == CTRL_PRIMARY_COMMITTING || role == CTRL_PRIMARY) {
/* Promoted connection dropped — remove from standby_endpoints */
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
for (int i = 0; i < ha->standby_endpoints->count; i++) {
if (zarray_get(ha->standby_endpoints, i) == ep) {
zarray_del(ha->standby_endpoints, i);
break;
}
}
pthread_mutex_unlock(&ha->lock);
}
} else {
/* Election phase: count as denied vote */
handle_request_vote_reply(ha, vctx->ctx, vctx->peer_idx,
vctx->my_term,
0, /* vote_granted = false */
0, 0);
}
/* Do NOT free vctx here — peer_ctrl_msg_handler may still reference ep->data.
* vctx is freed in ha_election_cleanup() when all per-peer contexts release. */
return 0;
}
@ -2443,17 +2603,7 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c,
n_peers);
/* Returns immediately — election result delivered via raft on_become_leader/follower */
}
/* ── Task 9.2: Persistent Control Channel ──────────────────────────────────
*
* ha_find_and_connect_to_primary: three-tier primary discovery + TCP connect.
*
* Requirements: 6.1
* Boundary: Persistent Control Channel (ha.c)
*/
#ifdef XCE_TESTING
int (*ha_tcp_connect_hook)(const char *host, int port) = NULL;
/*
* 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
@ -2465,158 +2615,6 @@ void (*ha_rdhup_election_hook)(ha_state_t *ha, cluster_t *c,
const char *my_name) = NULL;
#endif
/* ha_tcp_connect_raw: non-blocking connect + poll, returns connected fd or -errno */
static int ha_tcp_connect_raw(const char *host, int port, int timeout_ms)
{
#ifdef XCE_TESTING
if (ha_tcp_connect_hook)
return ha_tcp_connect_hook(host, port);
#endif
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), "%d", port);
if (getaddrinfo(host, port_str, &hints, &res) != 0)
return -EHOSTUNREACH;
int sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (sock < 0) { int e = errno; freeaddrinfo(res); return -e; }
int one = 1;
setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
int flags = fcntl(sock, F_GETFL, 0);
fcntl(sock, F_SETFL, flags | O_NONBLOCK);
connect(sock, res->ai_addr, res->ai_addrlen);
freeaddrinfo(res);
struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP };
int pr;
do { pr = poll(&pfd, 1, timeout_ms); } while (pr < 0 && errno == EINTR);
if (pr == 0) { close(sock); return -ETIMEDOUT; }
if (pr < 0 || (pfd.revents & (POLLERR | POLLHUP))) {
close(sock); return -ECONNREFUSED;
}
int err = 0; socklen_t el = sizeof(err);
getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
if (err) { close(sock); return -err; }
fcntl(sock, F_SETFL, flags & ~O_NONBLOCK);
return sock;
}
static int find_candidate_by_name(const xce_cluster_config_t *peers,
const char *name)
{
for (int i = 0; i < peers->candidates_count; i++) {
if (strcmp(peers->candidates[i].name, name) == 0)
return i;
}
return -1;
}
static void register_fd_with_iomgr(cluster_t *c, int fd)
{
if (!c) return;
struct epoll_event ev;
memset(&ev, 0, sizeof(ev));
ev.events = EPOLLIN | EPOLLRDHUP;
ev.data.fd = fd;
if (epoll_ctl(c->iomgr.epoll_fd, EPOLL_CTL_ADD, fd, &ev) != 0)
WARN("ha_find_and_connect: epoll_ctl: %s", strerror(errno));
}
static void ms_sleep_ha(long delay_ms)
{
if (delay_ms <= 0) return;
struct timespec ts = { .tv_sec = delay_ms / 1000,
.tv_nsec = (delay_ms % 1000) * 1000000L };
nanosleep(&ts, NULL);
}
int ha_find_and_connect_to_primary(ha_state_t *ha,
const ha_config_t *cfg,
const xce_cluster_config_t *peers,
cluster_t *c)
{
if (!ha || !cfg || !peers || peers->candidates_count <= 0)
return -EINVAL;
int timeout_ms = (int)cfg->probe_connect_timeout_ms;
if (timeout_ms <= 0)
timeout_ms = HA_DEFAULT_PROBE_CONNECT_TIMEOUT_MS;
/* Tier 1: voted_for_name → direct connect */
if (ha->voted_for_name[0] != '\0') {
int idx = find_candidate_by_name(peers, ha->voted_for_name);
if (idx >= 0) {
const xce_candidate_t *p1 = &peers->candidates[idx];
for (int r = 0; r < HA_FIND_PRIMARY_T1_MAX_RETRIES; r++) {
int fd = ha_tcp_connect_raw(p1->hostname, p1->port, timeout_ms);
if (fd >= 0) {
INFO("HA: Tier 1 connected to voted_for '%s' (retry=%d)",
ha->voted_for_name, r);
ha->primary_ctrl_fd = fd;
register_fd_with_iomgr(c, fd);
if (c) ha_arm_heartbeat_timeout_vtask(ha, c);
return fd;
}
if (fd == -ETIMEDOUT || fd == -EAGAIN) {
WARN("HA: Tier 1 '%s' not ready (retry %d/%d), backoff %dms",
ha->voted_for_name, r + 1,
HA_FIND_PRIMARY_T1_MAX_RETRIES, HA_FIND_PRIMARY_T1_BACKOFF_MS);
ms_sleep_ha(HA_FIND_PRIMARY_T1_BACKOFF_MS);
continue;
}
break; /* other errors → Tier 2 */
}
}
}
/* Tier 2: probe remaining candidates */
{
int t1_idx = find_candidate_by_name(peers, ha->voted_for_name);
for (int i = 0; i < peers->candidates_count; i++) {
if (i == t1_idx) continue;
const xce_candidate_t *p2 = &peers->candidates[i];
int fd = ha_tcp_connect_raw(p2->hostname, p2->port, timeout_ms);
if (fd >= 0) {
INFO("HA: Tier 2 connected to '%s'", p2->name);
ha->primary_ctrl_fd = fd;
register_fd_with_iomgr(c, fd);
if (c) ha_arm_heartbeat_timeout_vtask(ha, c);
return fd;
}
}
}
/* Tier 3: exponential backoff scan (no exit) */
INFO("HA: entering Tier 3 primary discovery");
long backoff_ms = HA_FIND_PRIMARY_T3_INIT_MS;
for (;;) {
for (int i = 0; i < peers->candidates_count; i++) {
const xce_candidate_t *p3 = &peers->candidates[i];
int fd = ha_tcp_connect_raw(p3->hostname, p3->port, timeout_ms);
if (fd >= 0) {
INFO("HA: Tier 3 connected to '%s' after backoff", p3->name);
ha->primary_ctrl_fd = fd;
register_fd_with_iomgr(c, fd);
if (c) ha_arm_heartbeat_timeout_vtask(ha, c);
return fd;
}
}
WARN("HA: Tier 3 — no primary, sleep %ldms", backoff_ms);
ms_sleep_ha(backoff_ms);
backoff_ms = backoff_ms * 2 > HA_FIND_PRIMARY_T3_MAX_MS
? HA_FIND_PRIMARY_T3_MAX_MS : backoff_ms * 2;
}
}
/* ── Task 9.4: EPOLLRDHUP Failover and Heartbeat Timeout ───────────────────
*
* Requirements: 6.2 (RDHUP fast path), 6.3 (heartbeat timeout fallback),
@ -2800,3 +2798,259 @@ void ha_disarm_heartbeat_timeout_vtask(ha_state_t *ha)
free(s_hb_vtask);
s_hb_vtask = NULL;
}
/* ── Primary heartbeat sender + late-joiner outbound connect ─────────────── */
/*
* standby_ep_ctx_t: per-peer context for outbound standby connections opened
* by the primary during the heartbeat task.
*
* Layout deliberately matches vote_ep_ctx_t for the first 4 fields so that
* peer_ctrl_msg_handler (which casts ep->data to vote_ep_ctx_t*) can safely
* read ha, c, and peer_idx. ctx is always NULL here (no active election).
*/
typedef struct {
ha_state_t *ha;
cluster_t *c;
ha_elect_ctx_t *ctx; /* always NULL — not a vote endpoint */
int peer_idx; /* index into c->myconf->peers_cfg.candidates[] */
} standby_ep_ctx_t;
/*
* on_standby_connected: iomgr on_connected callback for outbound standby
* endpoints created by the heartbeat task.
*
* Sends OPCode_hello (CtrlHelloRequest) so the receiving standby runs
* handle_ctrlhello and triggers WAL catch-up + config push. Registers the
* endpoint in ha->standby_endpoints so future heartbeats reach it.
*/
static void on_standby_connected(endpoint_t *ep)
{
standby_ep_ctx_t *sctx = (standby_ep_ctx_t *)ep->data;
ha_state_t *ha = sctx->ha;
cluster_t *c = sctx->c;
/* Register in standby_endpoints under lock (idempotency guard) */
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
int already = 0;
for (int i = 0; i < ha->standby_endpoints->count; i++) {
if (zarray_get(ha->standby_endpoints, i) == ep) { already = 1; break; }
}
if (!already)
zarray_store(ha->standby_endpoints, ep);
pthread_mutex_unlock(&ha->lock);
}
/* Send ctrlhello — triggers handle_ctrlhello on the standby side */
uint64_t lsn = atomic_load_explicit(&c->persist.committed_lsn,
memory_order_acquire);
uint64_t tag = c->myconf ? c->myconf->config_version_tag : 0;
char *name = c->myconf ? c->myconf->name : "";
ctrl_hello_req_t hello = { .name = name, .key = NULL,
.applied_lsn = lsn, .config_tag = tag };
xce_request_t req;
memset(&req, 0, sizeof(req));
req.opcode = OPCode_hello;
req.body_which = Request_body_ctrlhellomsg;
req.body.ctrlhellomsg = &hello;
int rc = xce_request(ep, &req);
if (rc < 0)
WARN("HA: ctrlhello to peer[%d] failed: %d", sctx->peer_idx, rc);
else
INFO("HA: outbound standby connection to peer[%d] established, ctrlhello sent",
sctx->peer_idx);
/* Set state-aware dispatcher for subsequent messages on this endpoint */
ep->msg_handler = peer_ctrl_msg_handler;
}
/*
* standby_ep_err_handler_primary: error handler for outbound standby
* endpoints created by the heartbeat task. Removes from standby_endpoints
* and frees the context — the next heartbeat tick will attempt reconnect.
*/
static int standby_ep_err_handler_primary(void *ep_raw)
{
endpoint_t *ep = (endpoint_t *)ep_raw;
standby_ep_ctx_t *sctx = (standby_ep_ctx_t *)ep->data;
if (!sctx) return 0;
ha_state_t *ha = sctx->ha;
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
for (int i = 0; i < ha->standby_endpoints->count; i++) {
if (zarray_get(ha->standby_endpoints, i) == ep) {
zarray_del(ha->standby_endpoints, i);
break;
}
}
pthread_mutex_unlock(&ha->lock);
}
free(sctx);
ep->data = NULL;
return 0;
}
/*
* ha_connect_to_peer: open a new outbound TLS connection to a peer candidate
* and register it as a standby channel. Called from primary_heartbeat_fn
* for peers that are absent from standby_endpoints.
*/
static void ha_connect_to_peer(cluster_t *c, int peer_idx)
{
ha_state_t *ha = &c->ha;
const xce_candidate_t *peer =
&c->myconf->peers_cfg.candidates[peer_idx];
standby_ep_ctx_t *sctx = calloc(1, sizeof(*sctx));
if (!sctx) return;
sctx->ha = ha;
sctx->c = c;
sctx->peer_idx = peer_idx;
endpoint_t *ep = endpoint_get();
if (!ep) { free(sctx); return; }
endpoint_init(ep, AF_INET, SOCK_STREAM, 0, 0);
endpoint_set_peer(ep, c->myhost);
ep->data = sctx;
ep->iomgr = &c->iomgr;
if (ha->client_ssl_ctx) {
SSL_CTX *client_ctx = (SSL_CTX *)ha->client_ssl_ctx;
ep->ssl = SSL_new(client_ctx);
if (ep->ssl)
SSL_set_connect_state(ep->ssl);
}
endpoint_set_on_connected(ep, on_standby_connected);
endpoint_set_err_handler(ep, standby_ep_err_handler_primary);
int rc = endpoint_connect(ep, peer->hostname, peer->port);
if (rc < 0) {
WARN("HA: connect to peer[%d] (%s:%u) failed: %d",
peer_idx, peer->hostname, peer->port, rc);
endpoint_fini(ep);
free(sctx);
return;
}
rc = iomgr_monitor(ep);
if (rc < 0) {
endpoint_fini(ep);
free(sctx);
return;
}
INFO("HA: connecting outbound to peer[%d] (%s:%u)",
peer_idx, peer->hostname, peer->port);
}
/*
* primary_heartbeat_fn: self-perpetuating one-shot delaymgr task.
*
* On each tick:
* 1. Sends OPCode_nop to all connected standbys (keeps watchdog alive).
* 2. Scans the peers list; for any peer NOT in standby_endpoints, opens a
* fresh outbound TLS connection so late joiners are bootstrapped without
* the standby ever initiating an outbound connect.
*
* Stops naturally on demotion: the role check prevents re-arming.
*/
static int primary_heartbeat_fn(void *arg)
{
cluster_t *c = (cluster_t *)arg;
ha_state_t *ha = &c->ha;
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
if (role != CTRL_PRIMARY && role != CTRL_PRIMARY_COMMITTING)
return 0; /* demoted — stop the heartbeat loop */
/* 1. Heartbeat all connected standbys */
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
int n = ha->standby_endpoints->count;
for (int i = 0; i < n; i++) {
endpoint_t *ep = (endpoint_t *)zarray_get(ha->standby_endpoints, i);
if (ep) {
int rc = xce_nop_req(ep);
if (rc < 0)
WARN("HA: heartbeat nop to standby[%d] failed: %d", i, rc);
}
}
pthread_mutex_unlock(&ha->lock);
}
/* 2. Reconnect to any peer missing from standby_endpoints */
if (c->myconf) {
const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
for (int i = 0; i < pcfg->candidates_count; i++) {
if (i == ha->my_priority) continue; /* skip self */
/* Check if this peer already has a standby endpoint.
*
* Two endpoint types can be in standby_endpoints:
* - Promoted vote eps: ep->data = vote_ep_ctx_t, peer_idx is
* votes_idx (relative, self excluded). Convert absolute i to
* votes_idx using the same formula as raft_cb_send_vote_request.
* - Heartbeat standby eps: ep->data = standby_ep_ctx_t (ctx==NULL),
* peer_idx is absolute candidate index. Compare directly with i.
*/
int votes_idx_for_i = (i < ha->my_priority) ? i : i - 1;
int found = 0;
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
for (int j = 0; j < ha->standby_endpoints->count; j++) {
endpoint_t *sep = (endpoint_t *)zarray_get(
ha->standby_endpoints, j);
if (!sep || !sep->data) continue;
standby_ep_ctx_t *sctx = (standby_ep_ctx_t *)sep->data;
int match_idx = (sctx->ctx != NULL)
? votes_idx_for_i /* promoted vote ep: relative */
: i; /* standby ep: absolute */
if (sctx->peer_idx == match_idx) { found = 1; break; }
}
pthread_mutex_unlock(&ha->lock);
}
if (!found) {
INFO("HA: peer[%d] (%s) not connected — initiating outbound connect",
i, pcfg->candidates[i].name ? pcfg->candidates[i].name : "?");
ha_connect_to_peer(c, i);
}
}
}
/* Re-arm for next interval — only reached when still PRIMARY */
int interval_ms = (int)ha->config.heartbeat_interval_ms;
if (interval_ms <= 0)
interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS;
delay_task_ms(&c->delaymgr, interval_ms, 1,
(delay_task_entry_fn_t)primary_heartbeat_fn, c);
return 0;
}
void ha_start_primary_heartbeat(cluster_t *c)
{
if (!c) return;
ha_state_t *ha = &c->ha;
int interval_ms = (int)ha->config.heartbeat_interval_ms;
if (interval_ms <= 0)
interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS;
/* Guard: skip if delaymgr is not running (e.g. unit tests with zeroed cluster_t).
* In production the delaymgr is always started before any promotion. */
if (!c->delaymgr.running)
return;
/* One-shot: the task re-arms itself each time while role == PRIMARY */
int rc = delay_task_ms(&c->delaymgr, interval_ms, 1,
(delay_task_entry_fn_t)primary_heartbeat_fn, c);
if (rc < 0)
WARN("ha_start_primary_heartbeat: delay_task_ms failed: %d", rc);
else
INFO("HA: primary heartbeat armed (interval=%dms)", interval_ms);
}

74
src/controller/ha.h
View file

@ -67,15 +67,19 @@ typedef enum {
* 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 */
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;
/*
@ -433,6 +437,16 @@ 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 ────────────────────── */
/*
@ -458,31 +472,6 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c,
uint64_t my_config_tag, uint64_t my_lsn,
const char *my_name);
/* ── Task 9.2: Persistent Control Channel (Req 6.1) ───────────────────────── */
/*
* ha_find_and_connect_to_primary: three-tier primary discovery and raw TCP connect.
*
* Tier 1: ha->voted_for_name non-empty → direct connect to that candidate.
* Backoff HA_FIND_PRIMARY_T1_BACKOFF_MS; retry HA_FIND_PRIMARY_T1_MAX_RETRIES.
* Tier 2: probe remaining candidates; follow XCE_ERR_STANDBY redirect.
* Tier 3: blind scan with 100ms→5s exponential backoff (no exit until success).
*
* On success: sets ha->primary_ctrl_fd; when c != NULL, registers fd with
* c->iomgr.epoll_fd for EPOLLIN|EPOLLRDHUP. Returns fd >= 0.
* Pass c == NULL to skip iomgr registration (unit test mode).
* Requirements: 6.1
*/
int ha_find_and_connect_to_primary(ha_state_t *ha,
const ha_config_t *cfg,
const xce_cluster_config_t *peers,
cluster_t *c);
#define HA_FIND_PRIMARY_T1_MAX_RETRIES 3
#define HA_FIND_PRIMARY_T1_BACKOFF_MS 200
#define HA_FIND_PRIMARY_T3_INIT_MS 100
#define HA_FIND_PRIMARY_T3_MAX_MS 5000
/*
* ha_repl_ack_received: called when a standby ACKs config+WAL catch-up
* (OPCode_replCatchUpDone). Increments repl_ack_count; when majority
@ -502,6 +491,7 @@ 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
@ -525,6 +515,17 @@ extern int (*ha_send_ctrlhello_hook)(void *c);
*/
extern int (*ha_push_config_hook)(void *c);
/*
* 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
@ -532,13 +533,6 @@ extern int (*ha_push_config_hook)(void *c);
*/
uint64_t ha_trigger_manual_failover_for_test(ha_state_t *ha);
/*
* ha_tcp_connect_hook: test-only seam for TCP connections.
* When non-NULL, ha_find_and_connect_to_primary() calls this instead of
* the real connect() path. Returns connected fd on success, -errno on failure.
*/
extern int (*ha_tcp_connect_hook)(const char *host, int port);
/*
* ha_rdhup_election_hook: test-only seam for ha_start_election_async in the
* RDHUP failover path.

2
src/controller/meson.build
View file

@ -108,7 +108,6 @@ controller_unit_test_src = [
'testsuite/test_ha_durability.c',
'testsuite/test_repl_majority.c',
'testsuite/test_persist_truncate.c',
'testsuite/test_ha_connect.c',
'testsuite/test_ha_ctrlhello.c',
'testsuite/test_ha_failover.c',
'testsuite/test_ha_vote_send.c',
@ -117,6 +116,7 @@ controller_unit_test_src = [
'testsuite/test_ha_evaluate_committing.c',
'testsuite/test_ha_repl_push.c',
'testsuite/test_ha_repl_ack.c',
'testsuite/test_ha_connection_reuse.c',
'startup_validation.c',
'worker.c',
'persist.c',

23
src/controller/myapp.c
View file

@ -252,29 +252,16 @@ int myapp_run(myconf_t *myconf)
skip_election:;
/* Nodes that skipped election (already voted or no config) are now STANDBY.
* Nodes that already have a config actively fetch it so they can connect to
* the primary. Nodes with no config (my_tag == 0) wait passively — the
* primary will push the config via ctrlhello after winning the election
* (Req 7.3). Triggering fetch_config_from_primary for no-config nodes
* only causes retry warnings since the primary isn't up yet. */
/* All standbys wait passively for the primary to connect outbound and push
* config via ctrlhello. fetch_config_from_primary has been removed:
* standbys never initiate outbound connects after an election result. */
{
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&c.ha.role,
memory_order_acquire);
uint64_t my_tag = c.myconf ? c.myconf->config_version_tag : 0;
if ((role == CTRL_STANDBY || role == CTRL_INIT) && my_tag > 0) {
if (role == CTRL_STANDBY || role == CTRL_INIT) {
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY,
memory_order_release);
extern int fetch_config_from_primary(cluster_t *c);
/* Use delaymgr one-shot so cluster_run is entered before the
* blocking fetch starts (fetch runs in delaymgr worker thread). */
delay_task_ms(&c.delaymgr, 200, 1,
(delay_task_entry_fn_t)fetch_config_from_primary, &c);
} else if (role == CTRL_STANDBY || role == CTRL_INIT) {
/* No config — stay STANDBY, wait for primary push (Req 7.3). */
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY,
memory_order_release);
INFO("myapp_run: no local config — waiting for primary to push config");
INFO("myapp_run: CTRL_STANDBY — waiting for primary to connect and push config");
}
}

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

@ -40,7 +40,6 @@ extern void test_ha_bootstrap(void);
extern void test_ha_durability(void);
extern void test_repl_majority(void);
extern void test_persist_truncate(void);
extern void test_ha_connect(void);
extern void test_ha_ctrlhello(void);
extern void test_ha_failover(void);
extern void test_ha_vote_send(void);
@ -49,6 +48,7 @@ extern void test_ha_vote_gate(void);
extern void test_ha_evaluate_committing(void);
extern void test_ha_repl_push(void);
extern void test_ha_repl_ack(void);
extern void test_ha_connection_reuse(void);
void setUp(void) { return; }
@ -158,7 +158,6 @@ int main(int argc, char* argv[])
TEST(test_ha_durability);
TEST(test_repl_majority);
TEST(test_persist_truncate);
TEST(test_ha_connect);
TEST(test_ha_ctrlhello);
TEST(test_ha_failover);
TEST(test_ha_vote_send);
@ -167,6 +166,7 @@ int main(int argc, char* argv[])
TEST(test_ha_evaluate_committing);
TEST(test_ha_repl_push);
TEST(test_ha_repl_ack);
TEST(test_ha_connection_reuse);
return UNITY_END();
}

230
src/controller/testsuite/test_ha_connect.c
View file

@ -1,230 +0,0 @@
/*
* test_ha_connect.c — Unity tests for ha_find_and_connect_to_primary() (task 9.2).
*
* Uses ha_tcp_connect_hook to inject mock fds; no real network connections.
* Requirements: 6.1
*/
#define _GNU_SOURCE
#include <errno.h>
#include <stdatomic.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/socket.h>
#include <unity/unity.h>
#include "ha.h"
/* ── Minimal ha_config_t for tests ─────────────────────────────── */
static ha_config_t g_cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 500,
.probe_connect_timeout_ms = 50,
.max_promotion_retries = 3,
.election_jitter_max_ms = 0,
.election_priority_step_ms = 500,
};
/* ── Mock connect hook state ────────────────────────────────────── */
static const char *g_last_host = NULL;
static int g_last_port = 0;
static int g_mock_fd = -1;
static int g_fail_err = 0;
static int g_call_count = 0;
static int mock_connect(const char *host, int port)
{
g_call_count++;
g_last_host = host;
g_last_port = port;
if (g_fail_err) return -g_fail_err;
return g_mock_fd;
}
static void reset_mock(int mock_fd, int fail_err)
{
g_last_host = NULL;
g_last_port = 0;
g_mock_fd = mock_fd;
g_fail_err = fail_err;
g_call_count = 0;
ha_tcp_connect_hook = mock_connect;
}
/* ── Helper: build a small peers config ────────────────────────── */
static void build_peers(xce_cluster_config_t *peers, int n,
const char *names[], const char *hosts[], int ports[])
{
memset(peers, 0, sizeof(*peers));
peers->candidates_count = n;
for (int i = 0; i < n; i++) {
strncpy(peers->candidates[i].name, names[i],
sizeof(peers->candidates[i].name) - 1);
strncpy(peers->candidates[i].hostname, hosts[i],
sizeof(peers->candidates[i].hostname) - 1);
peers->candidates[i].port = ports[i];
}
}
/* ── Test: primary_ctrl_fd initialised to -1 ───────────────────── */
static void test_primary_ctrl_fd_init_minus_one(void)
{
ha_state_t ha;
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &g_cfg, 1));
TEST_ASSERT_EQUAL_INT(-1, ha.primary_ctrl_fd);
ha_fini(&ha);
printf("test_primary_ctrl_fd_init_minus_one: PASS\n");
}
/* ── Test: NULL peers → -EINVAL ─────────────────────────────────── */
static void test_null_peers_returns_einval(void)
{
ha_state_t ha;
ha_init(&ha, &g_cfg, 1);
int rc = ha_find_and_connect_to_primary(&ha, &g_cfg, NULL, NULL);
TEST_ASSERT_LESS_THAN_INT(0, rc);
ha_fini(&ha);
printf("test_null_peers_returns_einval: PASS\n");
}
/* ── Test: empty peers list → -EINVAL ────────────────────────────── */
static void test_empty_peers_returns_einval(void)
{
ha_state_t ha;
ha_init(&ha, &g_cfg, 1);
xce_cluster_config_t peers;
memset(&peers, 0, sizeof(peers));
peers.candidates_count = 0;
int rc = ha_find_and_connect_to_primary(&ha, &g_cfg, &peers, NULL);
TEST_ASSERT_LESS_THAN_INT(0, rc);
ha_fini(&ha);
printf("test_empty_peers_returns_einval: PASS\n");
}
/* ── Test: voted_for_name → Tier 1 connect attempted first ──────── */
static void test_tier1_voted_for_name_tried_first(void)
{
int sv[2];
TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv));
reset_mock(sv[0], 0);
ha_state_t ha;
ha_init(&ha, &g_cfg, 1);
strncpy(ha.voted_for_name, "ctrl-primary", sizeof(ha.voted_for_name) - 1);
const char *names[] = { "ctrl-primary", "ctrl-b" };
const char *hosts[] = { "10.0.0.1", "10.0.0.2" };
int ports[] = { 9990, 9990 };
xce_cluster_config_t peers;
build_peers(&peers, 2, names, hosts, ports);
int fd = ha_find_and_connect_to_primary(&ha, &g_cfg, &peers, NULL);
TEST_ASSERT_EQUAL_INT(sv[0], fd);
TEST_ASSERT_EQUAL_INT(sv[0], ha.primary_ctrl_fd);
TEST_ASSERT_EQUAL_STRING("10.0.0.1", g_last_host);
TEST_ASSERT_EQUAL_INT(9990, g_last_port);
TEST_ASSERT_GREATER_THAN_INT(0, g_call_count);
close(sv[1]);
ha.primary_ctrl_fd = -1;
close(sv[0]);
ha_fini(&ha);
printf("test_tier1_voted_for_name_tried_first: PASS\n");
}
/* ── Test: voted_for_name absent → Tier 2 scan ───────────────────── */
static void test_tier2_scan_when_voted_for_missing(void)
{
int sv[2];
TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv));
reset_mock(sv[0], 0);
ha_state_t ha;
ha_init(&ha, &g_cfg, 1);
strncpy(ha.voted_for_name, "ctrl-missing", sizeof(ha.voted_for_name) - 1);
const char *names[] = { "ctrl-a", "ctrl-b" };
const char *hosts[] = { "10.0.0.1", "10.0.0.2" };
int ports[] = { 9990, 9990 };
xce_cluster_config_t peers;
build_peers(&peers, 2, names, hosts, ports);
int fd = ha_find_and_connect_to_primary(&ha, &g_cfg, &peers, NULL);
TEST_ASSERT_EQUAL_INT(sv[0], fd);
TEST_ASSERT_EQUAL_INT(sv[0], ha.primary_ctrl_fd);
close(sv[1]);
ha.primary_ctrl_fd = -1;
close(sv[0]);
ha_fini(&ha);
printf("test_tier2_scan_when_voted_for_missing: PASS\n");
}
/* ── Test: successful connect sets primary_ctrl_fd ──────────────── */
static void test_successful_connect_sets_primary_ctrl_fd(void)
{
int sv[2];
TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv));
reset_mock(sv[0], 0);
ha_state_t ha;
ha_init(&ha, &g_cfg, 0);
strncpy(ha.voted_for_name, "ctrl-a", sizeof(ha.voted_for_name) - 1);
const char *names[] = { "ctrl-a" };
const char *hosts[] = { "127.0.0.1" };
int ports[] = { 9990 };
xce_cluster_config_t peers;
build_peers(&peers, 1, names, hosts, ports);
int fd = ha_find_and_connect_to_primary(&ha, &g_cfg, &peers, NULL);
TEST_ASSERT_EQUAL_INT(sv[0], fd);
TEST_ASSERT_EQUAL_INT(sv[0], ha.primary_ctrl_fd);
close(sv[1]);
ha.primary_ctrl_fd = -1;
close(sv[0]);
ha_fini(&ha);
printf("test_successful_connect_sets_primary_ctrl_fd: PASS\n");
}
/* ── Test: function symbol is resolvable ─────────────────────────── */
static void test_ha_find_and_connect_symbol_exists(void)
{
typedef int (*fn_t)(ha_state_t *, const ha_config_t *,
const xce_cluster_config_t *, cluster_t *);
fn_t fn = ha_find_and_connect_to_primary;
TEST_ASSERT_NOT_NULL((void *)fn);
printf("test_ha_find_and_connect_symbol_exists: PASS\n");
}
/* ── Suite entry point ───────────────────────────────────────────── */
void test_ha_connect(void)
{
ha_tcp_connect_hook = mock_connect;
UNITY_BEGIN();
RUN_TEST(test_primary_ctrl_fd_init_minus_one);
RUN_TEST(test_null_peers_returns_einval);
RUN_TEST(test_empty_peers_returns_einval);
RUN_TEST(test_tier1_voted_for_name_tried_first);
RUN_TEST(test_tier2_scan_when_voted_for_missing);
RUN_TEST(test_successful_connect_sets_primary_ctrl_fd);
RUN_TEST(test_ha_find_and_connect_symbol_exists);
UNITY_END();
}

1
src/include/request.h
View file

@ -19,6 +19,7 @@ typedef struct {
typedef int (*xce_check_fn_t)(void*);
extern int xce_request(endpoint_t *ep, xce_request_t *req);
extern int xce_encode(xce_request_t*, io_packet_t*);
extern int xce_decode(io_packet_t*, xce_request_t*);