From 7b396502291053ce7d5677fa21b110d4a166915e Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Mon, 1 Jun 2026 04:17:05 +0000 Subject: [PATCH] =?UTF-8?q?refactor(ha):=20passive=20standby=20model=20?= =?UTF-8?q?=E2=80=94=20primary=20connects=20outbound=20to=20all=20peers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- CLAUDE.md | 29 + src/controller/CLAUDE.md | 36 + src/controller/cluster.c | 8 +- src/controller/config_repl.c | 436 +++--------- src/controller/ha.c | 792 ++++++++++++++------- src/controller/ha.h | 74 +- src/controller/meson.build | 2 +- src/controller/myapp.c | 23 +- src/controller/testsuite/controller_test.c | 4 +- src/controller/testsuite/test_ha_connect.c | 230 ------ src/include/request.h | 1 + 11 files changed, 716 insertions(+), 919 deletions(-) delete mode 100644 src/controller/testsuite/test_ha_connect.c diff --git a/CLAUDE.md b/CLAUDE.md index 6b5ddbf1..04636af0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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 diff --git a/src/controller/CLAUDE.md b/src/controller/CLAUDE.md index 7cb5dffe..e9227d34 100644 --- a/src/controller/CLAUDE.md +++ b/src/controller/CLAUDE.md @@ -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 diff --git a/src/controller/cluster.c b/src/controller/cluster.c index c736126f..11d3d6ee 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -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. diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 76c60bc3..f3f9c461 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -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: / */ - 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); diff --git a/src/controller/ha.c b/src/controller/ha.c index 326de6d3..35deaefb 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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); +} + diff --git a/src/controller/ha.h b/src/controller/ha.h index ae298db1..daa8dace 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -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. diff --git a/src/controller/meson.build b/src/controller/meson.build index f5f69775..98ca7687 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -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', diff --git a/src/controller/myapp.c b/src/controller/myapp.c index 32dc979b..36026341 100644 --- a/src/controller/myapp.c +++ b/src/controller/myapp.c @@ -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"); } } diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 0c6ec1df..58f3d358 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -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(); } diff --git a/src/controller/testsuite/test_ha_connect.c b/src/controller/testsuite/test_ha_connect.c deleted file mode 100644 index c782c149..00000000 --- a/src/controller/testsuite/test_ha_connect.c +++ /dev/null @@ -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 -#include -#include -#include -#include -#include -#include -#include - -#include -#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(); -} diff --git a/src/include/request.h b/src/include/request.h index 2eed8fa2..bf219ead 100644 --- a/src/include/request.h +++ b/src/include/request.h @@ -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*);