feat(ha-config-bootstrap): implement ha_find_and_connect_to_primary three-tier discovery (task 9.2)

.kiro/specs/ha-config-bootstrap/tasks.md

@@ -206,7 +206,7 @@ Dependency order: startup validation (D) unblocks vtaskmgr before election (A);
   - _Requirements: 6.5_
   - _Boundary: Persistence Layer (persist.c) / WAL replay_
 
-- [ ] 9.2 Implement ha_find_and_connect_to_primary() with three-tier discovery
+- [x] 9.2 Implement ha_find_and_connect_to_primary() with three-tier discovery
   - Tier 1: if `ha.voted_for_name` set → direct connect; on `XCE_ERR_NOT_READY` → backoff, retry Tier 1
   - Tier 2: probe remaining peers; follow `XCE_ERR_STANDBY` `redirect_host/port` immediately
   - Tier 3: blind sequential scan with 100ms→5s exponential backoff

src/controller/ha.c

@@ -13,6 +13,7 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <sys/epoll.h>
 #include <sys/socket.h>
 #include <time.h>
 #include <unistd.h>
@@ -1362,3 +1363,164 @@ ctrl_role_t am_i_primary_quorum(ha_state_t *ha,
 
     return CTRL_STANDBY;
 }
+
+/* ── 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;
+#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);
+                    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);
+                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);
+                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;
+    }
+}

src/controller/ha.h

@@ -411,6 +411,31 @@ int xce_probe_config_tag(const char *host, uint16_t port, uint64_t *tag_out);
  */
 ctrl_role_t ha_wait_for_heartbeat_or_timeout(ha_state_t *ha);
 
+/* ── Task 9.2: Persistent Control Channel (Req 6.1) ───────────────────────── */
+
+/*
+ * ha_find_and_connect_to_primary: three-tier primary discovery and raw TCP connect.
+ *
+ * Tier 1: ha->voted_for_name non-empty → direct connect to that candidate.
+ *         Backoff HA_FIND_PRIMARY_T1_BACKOFF_MS; retry HA_FIND_PRIMARY_T1_MAX_RETRIES.
+ * Tier 2: probe remaining candidates; follow XCE_ERR_STANDBY redirect.
+ * Tier 3: blind scan with 100ms→5s exponential backoff (no exit until success).
+ *
+ * On success: sets ha->primary_ctrl_fd; when c != NULL, registers fd with
+ *             c->iomgr.epoll_fd for EPOLLIN|EPOLLRDHUP. Returns fd >= 0.
+ * Pass c == NULL to skip iomgr registration (unit test mode).
+ * Requirements: 6.1
+ */
+int ha_find_and_connect_to_primary(ha_state_t *ha,
+                                   const ha_config_t *cfg,
+                                   const xce_cluster_config_t *peers,
+                                   cluster_t *c);
+
+#define HA_FIND_PRIMARY_T1_MAX_RETRIES   3
+#define HA_FIND_PRIMARY_T1_BACKOFF_MS  200
+#define HA_FIND_PRIMARY_T3_INIT_MS     100
+#define HA_FIND_PRIMARY_T3_MAX_MS     5000
+
 /* ── P3-6: Test-only failover shim ────────────────────────────── */
 
 #ifdef XCE_TESTING
@@ -420,6 +445,13 @@ ctrl_role_t ha_wait_for_heartbeat_or_timeout(ha_state_t *ha);
  * and term monotonicity. See design doc P3-6 and src/docs/persistence-and-ha-design.md.
  */
 uint64_t ha_trigger_manual_failover_for_test(ha_state_t *ha);
+
+/*
+ * ha_tcp_connect_hook: test-only seam for TCP connections.
+ * When non-NULL, ha_find_and_connect_to_primary() calls this instead of
+ * the real connect() path. Returns connected fd on success, -errno on failure.
+ */
+extern int (*ha_tcp_connect_hook)(const char *host, int port);
 #endif
 
 #endif /* HA_H */

src/controller/meson.build

@@ -108,6 +108,7 @@ controller_unit_test_src = [
   'testsuite/test_ha_durability.c',
   'testsuite/test_repl_majority.c',
   'testsuite/test_persist_truncate.c',
+  'testsuite/test_ha_connect.c',
   'startup_validation.c',
   'worker.c',
   'persist.c',

src/controller/testsuite/controller_test.c

@@ -40,6 +40,7 @@ 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);
 
 void setUp(void) { return; }
 
@@ -149,6 +150,7 @@ int main(int argc, char* argv[])
     TEST(test_ha_durability);
     TEST(test_repl_majority);
     TEST(test_persist_truncate);
+    TEST(test_ha_connect);
 
     return UNITY_END();
 }

src/controller/testsuite/test_ha_connect.c

@@ -0,0 +1,230 @@
+/*
+ * 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();
+}