feat: implement real am_i_primary() with election probing, ha_is_primary(), ha_check_priority_stepdown()

- Replace stub am_i_primary() with real election probe logic
- probe_thread(): non-blocking TCP connect with optional SSL handshake
- wait_any_probe(): deadline-based spin-poll across all probe threads
- am_i_primary(): priority-0 always primary; higher priorities probe
  lower-priority peers and self-promote if deadline expires
- ha_is_primary(): lock-free role check
- ha_check_priority_stepdown(): simultaneous-boot stepdown at term=0
- start/stop_heartbeat_listener(): stubs for Task 7
- ha_init(): explicit initialization of ssl_ctx/jitter_seed/standby_endpoints
- TDD: add test_am_i_primary suite with 3 tests covering p0/p1 paths

src/controller/ha.c

@@ -1,21 +1,29 @@
 #define _GNU_SOURCE
 #include "ha.h"
 
+#include <arpa/inet.h>
 #include <errno.h>
 #include <fcntl.h>
+#include <netdb.h>
+#include <poll.h>
 #include <pthread.h>
 #include <signal.h>
 #include <stdatomic.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <sys/socket.h>
 #include <time.h>
 #include <unistd.h>
 
+#include <openssl/ssl.h>
+#include <openssl/err.h>
+
 /* Circular-include safety: cluster_def.h includes ha.h (include-guarded),
  * and ha.h only forward-declares cluster_t — no cycle. */
 #include "xce/cluster_def.h"
 #include "persist.h"
+#include "utils.h"
 
 /* ── PRNG seeding ───────────────────────────────────────────── */
 
@@ -69,6 +77,11 @@ int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority)
     atomic_store_explicit(&ha->watchdog_stop,         0,    memory_order_relaxed);
 
     pthread_mutex_init(&ha->lock, NULL);
+
+    ha->ssl_ctx            = NULL;
+    ha->jitter_seed        = 0;
+    ha->standby_endpoints  = NULL;
+
     return 0;
 }
 
@@ -114,6 +127,231 @@ void ha_do_demote(ha_state_t *ha)
     /* Hook disable requires hooksys_t — done by ha_demote_to_standby(cluster_t*) */
 }
 
+/* ── P1-6: Election probing ─────────────────────────────────── */
+
+/*
+ * probe_thread: attempt a TCP (optionally SSL) connection to one candidate.
+ * Sets ctx->got_ack = 1 if the connection succeeds before election_done is set.
+ */
+static void *probe_thread(void *arg)
+{
+    probe_ctx_t *ctx = (probe_ctx_t *)arg;
+
+    struct addrinfo hints, *res = NULL;
+    memset(&hints, 0, sizeof(hints));
+    hints.ai_family   = AF_UNSPEC;
+    hints.ai_socktype = SOCK_STREAM;
+
+    char port_str[16];
+    snprintf(port_str, sizeof(port_str), "%u", (unsigned)ctx->candidate->port);
+    if (getaddrinfo(ctx->candidate->hostname, port_str, &hints, &res) != 0)
+        goto done;
+
+    int sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
+    if (sock < 0) { freeaddrinfo(res); goto done; }
+
+    /* Non-blocking connect */
+    int flags = fcntl(sock, F_GETFL, 0);
+    fcntl(sock, F_SETFL, flags | O_NONBLOCK);
+    connect(sock, res->ai_addr, res->ai_addrlen);
+    freeaddrinfo(res);
+    res = NULL;
+
+    /* Poll loop: check election_done between iterations */
+    while (!atomic_load(ctx->election_done)) {
+        struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP };
+        int pr = poll(&pfd, 1, (int)ctx->timeout_ms);
+        if (pr < 0 && errno == EINTR) continue;
+        if (pr == 0) goto cleanup; /* poll timeout */
+        if (pfd.revents & (POLLERR | POLLHUP)) goto cleanup;
+        if (pfd.revents & POLLOUT) break;
+    }
+    if (atomic_load(ctx->election_done)) goto cleanup;
+
+    /* Verify connect succeeded */
+    {
+        int err = 0;
+        socklen_t el = sizeof(err);
+        getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
+        if (err != 0) goto cleanup;
+    }
+
+    /* SSL handshake — only if ssl_ctx is set */
+    if (ctx->ssl_ctx) {
+        SSL *ssl = SSL_new((SSL_CTX *)ctx->ssl_ctx);
+        if (!ssl) goto cleanup;
+        /* Restore blocking for SSL */
+        fcntl(sock, F_SETFL, flags & ~O_NONBLOCK);
+        SSL_set_fd(ssl, sock);
+        int sr;
+        do {
+            sr = SSL_connect(ssl);
+        } while (sr <= 0 &&
+                 (SSL_get_error(ssl, sr) == SSL_ERROR_WANT_READ ||
+                  SSL_get_error(ssl, sr) == SSL_ERROR_WANT_WRITE) &&
+                 !atomic_load(ctx->election_done));
+        SSL_shutdown(ssl);
+        SSL_free(ssl);
+        if (sr <= 0 || atomic_load(ctx->election_done))
+            goto cleanup;
+    }
+
+    ctx->got_ack = 1;
+
+cleanup:
+    close(sock);
+done:
+    return NULL;
+}
+
+/*
+ * wait_any_probe: spin-poll until any probe sets got_ack, or deadline passes.
+ * Returns 1 if any probe acknowledged, 0 if deadline expired.
+ */
+static int wait_any_probe(probe_ctx_t *probes, int n,
+                          const struct timespec *deadline)
+{
+    while (1) {
+        for (int i = 0; i < n; i++) {
+            if (probes[i].got_ack)
+                return 1;
+        }
+        struct timespec now;
+        clock_gettime(CLOCK_MONOTONIC, &now);
+        if (now.tv_sec > deadline->tv_sec ||
+            (now.tv_sec == deadline->tv_sec &&
+             now.tv_nsec >= deadline->tv_nsec))
+            return 0;
+        /* Sleep 10ms between checks */
+        struct timespec sleep_ts = { .tv_sec = 0, .tv_nsec = 10 * 1000000L };
+        nanosleep(&sleep_ts, NULL);
+    }
+}
+
+/*
+ * am_i_primary: election probe logic.
+ *
+ * Priority-0 is unconditionally primary — never probes.
+ * Priority-N probes the N candidates with lower index (higher priority).
+ * If any responds before election_timeout + priority_step*N + jitter, this
+ * node becomes standby.  If deadline expires with no response, self-promote.
+ */
+int am_i_primary(cluster_t *c)
+{
+    /* Priority-0 never probes — it is unconditionally primary */
+    if (c->ha.my_priority == 0)
+        return 1;
+
+    const xce_cluster_config_t *pcfg = &c->myconf->peers_cfg;
+    int n_probes = (c->ha.my_priority < pcfg->candidates_count)
+                       ? c->ha.my_priority
+                       : pcfg->candidates_count;
+    if (n_probes == 0)
+        return 1;
+
+    /* Seed jitter PRNG if needed */
+    if (c->ha.jitter_seed == 0)
+        ha_seed_jitter(&c->ha.jitter_seed, c->ha.my_priority);
+
+    atomic_int election_done;
+    atomic_init(&election_done, 0);
+    probe_ctx_t probes[MAX_HA_CANDIDATES];
+
+    for (int i = 0; i < n_probes; i++) {
+        probes[i].candidate     = (xce_candidate_t *)&pcfg->candidates[i];
+        probes[i].ssl_ctx       = c->ha.ssl_ctx;
+        probes[i].timeout_ms    = c->ha.config.probe_connect_timeout_ms;
+        probes[i].election_done = &election_done;
+        probes[i].got_ack       = 0;
+        pthread_create(&probes[i].thread, NULL, probe_thread, &probes[i]);
+    }
+
+    /* Compute deadline = election_timeout + priority_step * my_priority + jitter */
+    long to_ms = (long)c->ha.config.election_timeout_ms
+               + (long)c->ha.my_priority
+                 * (long)c->ha.config.election_priority_step_ms;
+    if (c->ha.config.election_jitter_max_ms > 0) {
+        to_ms += (long)(xorshift64_next(&c->ha.jitter_seed)
+                        % ((uint64_t)c->ha.config.election_jitter_max_ms + 1));
+    }
+
+    struct timespec deadline;
+    clock_gettime(CLOCK_MONOTONIC, &deadline);
+    deadline.tv_sec  += to_ms / 1000;
+    deadline.tv_nsec += (to_ms % 1000) * 1000000L;
+    if (deadline.tv_nsec >= 1000000000L) {
+        deadline.tv_sec++;
+        deadline.tv_nsec -= 1000000000L;
+    }
+
+    int got_ack = wait_any_probe(probes, n_probes, &deadline);
+
+    /* Signal all threads to stop */
+    atomic_store(&election_done, 1);
+    for (int i = 0; i < n_probes; i++)
+        pthread_join(probes[i].thread, NULL);
+
+    if (got_ack) {
+        WARN("HA election: higher-priority candidate responded — "
+             "starting as standby (my_priority=%d)", c->ha.my_priority);
+        return 0;
+    }
+    INFO("HA election: no higher-priority candidate responded within %ldms — "
+         "starting as primary (my_priority=%d)", to_ms, c->ha.my_priority);
+    return 1;
+}
+
+/*
+ * ha_is_primary: lock-free role check.
+ */
+int ha_is_primary(const ha_state_t *ha)
+{
+    return (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire)
+           == CTRL_PRIMARY;
+}
+
+/*
+ * ha_check_priority_stepdown: called from heartbeat handler on simultaneous boot
+ * (both nodes at term=0). Steps down PRIMARY to STANDBY if a higher-priority peer
+ * (lower index = lower sender_priority value) is observed.
+ */
+void ha_check_priority_stepdown(ha_state_t *ha, int sender_priority,
+                                uint64_t sender_term)
+{
+    /* Only act if sender has strictly higher priority (lower index) */
+    if (sender_priority >= ha->my_priority)
+        return;
+
+    uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire);
+    if (my_term > 0 || sender_term > 0)
+        return; /* term-based election took over — don't interfere */
+
+    ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
+                                                          memory_order_acquire);
+    if (role == CTRL_PRIMARY) {
+        WARN("HA: stepping down — higher-priority peer (priority=%d) present at term=0",
+             sender_priority);
+        ha_do_demote(ha);
+    }
+}
+
+/*
+ * start_heartbeat_listener / stop_heartbeat_listener:
+ * Stub implementations — full listener is implemented in Task 7 (cluster.c).
+ * The stubs allow am_i_primary() callers (main.c / cluster.c) to compile and
+ * link without requiring the full Phase-2 network stack.
+ */
+void start_heartbeat_listener(cluster_t *c)
+{
+    (void)c;
+    /* Heartbeat listener stub — full implementation in cluster.c Task 7 */
+}
+
+void stop_heartbeat_listener(cluster_t *c)
+{
+    (void)c;
+}
+
 /* ── P3-6: Test-only failover shim ────────────────────────────── */
 
 #ifdef XCE_TESTING
@@ -238,12 +476,6 @@ int ha_no_primary_known(const ha_state_t *ha)
     return (ha->primary_host[0] == '\0' || ha->primary_port == 0) ? 1 : 0;
 }
 
-/* priority-0 node is primary; full election probing is a future enhancement */
-int am_i_primary(cluster_t *c)
-{
-    return c->ha.my_priority == 0;
-}
-
 /*
  * ha_promote_to_primary: increment term, set role=PRIMARY, then register
  * persistence hooks (idempotency guard via hooks_registered).

src/controller/testsuite/controller_test.c

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

src/controller/testsuite/test_ha_role.c

@@ -7,6 +7,8 @@
 #include <unity/unity.h>
 
 #include "ha.h"
+#include "myconf.h"
+#include "xce/cluster_def.h"
 
 static void test_ha_init_default_role_is_standby(void)
 {
@@ -232,6 +234,92 @@ static void test_sigusr1_sets_failover_requested(void)
     printf("test_sigusr1_sets_failover_requested: PASS\n");
 }
 
+/* ── P1-6: am_i_primary() election tests ────────────────────────────── */
+
+/* test: priority-0 with no peers → immediately primary (no probing needed) */
+void test_am_i_primary_p0_no_peers(void)
+{
+    cluster_t c;
+    memset(&c, 0, sizeof(c));
+    myconf_t myconf;
+    memset(&myconf, 0, sizeof(myconf));
+    c.myconf = &myconf;
+
+    ha_config_t cfg = {
+        .heartbeat_interval_ms     = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
+        .heartbeat_timeout_ms      = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
+        .election_timeout_ms       = 500,  /* minimum valid value */
+        .probe_connect_timeout_ms  = 100,
+        .max_promotion_retries     = 1,
+        .election_jitter_max_ms    = 0,
+        .election_priority_step_ms = 0,
+    };
+    TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &cfg, 0));
+    TEST_ASSERT_EQUAL_INT(1, am_i_primary(&c));
+    ha_fini(&c.ha);
+    printf("test_am_i_primary_p0_no_peers: PASS\n");
+}
+
+/* test: priority-0 with peers → still primary (never probes higher-priority nodes) */
+void test_am_i_primary_p0_with_peers(void)
+{
+    cluster_t c;
+    memset(&c, 0, sizeof(c));
+    myconf_t myconf;
+    memset(&myconf, 0, sizeof(myconf));
+    myconf.peers_cfg.candidates_count = 2;
+    strncpy(myconf.peers_cfg.candidates[0].hostname, "127.0.0.1", XCE_MAXNAME_LEN - 1);
+    myconf.peers_cfg.candidates[0].port = 19900;
+    strncpy(myconf.peers_cfg.candidates[1].hostname, "127.0.0.1", XCE_MAXNAME_LEN - 1);
+    myconf.peers_cfg.candidates[1].port = 19901;
+    c.myconf = &myconf;
+
+    ha_config_t cfg = {
+        .heartbeat_interval_ms     = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
+        .heartbeat_timeout_ms      = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
+        .election_timeout_ms       = 500,  /* minimum valid value */
+        .probe_connect_timeout_ms  = 100,
+        .max_promotion_retries     = 1,
+        .election_jitter_max_ms    = 0,
+        .election_priority_step_ms = 0,
+    };
+    TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &cfg, 0));
+    TEST_ASSERT_EQUAL_INT(1, am_i_primary(&c));
+    ha_fini(&c.ha);
+    printf("test_am_i_primary_p0_with_peers: PASS\n");
+}
+
+/* test: priority-1 with no higher-priority peer alive → deadline expires → primary */
+void test_am_i_primary_p1_deadline_expires(void)
+{
+    cluster_t c;
+    memset(&c, 0, sizeof(c));
+    myconf_t myconf;
+    memset(&myconf, 0, sizeof(myconf));
+    /* Point to ports where nothing is listening */
+    myconf.peers_cfg.candidates_count = 2;
+    strncpy(myconf.peers_cfg.candidates[0].hostname, "127.0.0.1", XCE_MAXNAME_LEN - 1);
+    myconf.peers_cfg.candidates[0].port = 19910; /* nothing listening */
+    strncpy(myconf.peers_cfg.candidates[1].hostname, "127.0.0.1", XCE_MAXNAME_LEN - 1);
+    myconf.peers_cfg.candidates[1].port = 19911;
+    c.myconf = &myconf;
+
+    ha_config_t cfg = {
+        .heartbeat_interval_ms     = HA_DEFAULT_HEARTBEAT_INTERVAL_MS,
+        .heartbeat_timeout_ms      = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS,
+        .election_timeout_ms       = 600,  /* expires quickly; > 500ms floor */
+        .probe_connect_timeout_ms  = 200,
+        .max_promotion_retries     = 1,
+        .election_jitter_max_ms    = 0,
+        .election_priority_step_ms = 0,
+    };
+    TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &cfg, 1)); /* priority 1 — must probe priority 0 */
+    int result = am_i_primary(&c);
+    TEST_ASSERT_EQUAL_INT(1, result); /* no peer responded → self-declare primary */
+    ha_fini(&c.ha);
+    printf("test_am_i_primary_p1_deadline_expires: PASS\n");
+}
+
 void test_ha_role(void)
 {
     RUN_TEST(test_ha_init_default_role_is_standby);
@@ -247,3 +335,12 @@ void test_ha_role(void)
     RUN_TEST(test_demote_sets_standby_role);
     RUN_TEST(test_sigusr1_sets_failover_requested);
 }
+
+/* ── P1-6 suite: am_i_primary() election ─────────────────────────────── */
+
+void test_am_i_primary(void)
+{
+    RUN_TEST(test_am_i_primary_p0_no_peers);
+    RUN_TEST(test_am_i_primary_p0_with_peers);
+    RUN_TEST(test_am_i_primary_p1_deadline_expires);
+}