diff --git a/.kiro/specs/ha-config-bootstrap/tasks.md b/.kiro/specs/ha-config-bootstrap/tasks.md index 05e6606d..4157c5f6 100644 --- a/.kiro/specs/ha-config-bootstrap/tasks.md +++ b/.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 diff --git a/src/controller/ha.c b/src/controller/ha.c index b1bcb57a..26adf3a4 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -13,6 +13,7 @@ #include #include #include +#include #include #include #include @@ -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; + } +} diff --git a/src/controller/ha.h b/src/controller/ha.h index cd57d559..e48d40ed 100644 --- a/src/controller/ha.h +++ b/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 */ diff --git a/src/controller/meson.build b/src/controller/meson.build index 0f1125e5..f7c03be8 100644 --- a/src/controller/meson.build +++ b/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', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 21c597f7..8c9c2c8b 100644 --- a/src/controller/testsuite/controller_test.c +++ b/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(); } diff --git a/src/controller/testsuite/test_ha_connect.c b/src/controller/testsuite/test_ha_connect.c new file mode 100644 index 00000000..c782c149 --- /dev/null +++ b/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 +#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(); +}