myxce/src/lib/cluster.c

/* src/lib/cluster.c — cluster discovery and reconnect (P2.5-2/3)
 *
 * xce_connect_cluster() iterates the candidate list and returns the first
 * successful endpoint_t* (cast to void*).  In production the real SSL
 * connect is performed via do_connect(); unit tests override s_connect_fn.
 *
 * xce_reconnect_step() drives a non-blocking state machine:
 *   IDLE → CONNECTING → CONNECTED  (happy path)
 *   IDLE → CONNECTING → BACKOFF → CONNECTING → …  (retrying)
 *
 * Backoff sleep is skipped when s_connect_fn is set (i.e., in unit tests)
 * so tests do not need to wait for real timeouts.
 */
#include <errno.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#include "xce/cluster.h"

/* ── Peer list parsing ─────────────────────────────────────────── */

int xce_parse_peers(const char* peers_str, xce_cluster_config_t* cfg)
{
    cfg->candidates_count = 0;
    if (!peers_str || peers_str[0] == '\0')
        return -1;

    char buf[2048];
    strncpy(buf, peers_str, sizeof(buf) - 1);
    buf[sizeof(buf) - 1] = '\0';

    char* saveptr;
    char* tok = strtok_r(buf, ",", &saveptr);
    while (tok && cfg->candidates_count < MAX_HA_CANDIDATES) {
        /* Format: name=host:port  OR  host:port */
        char* eq    = strchr(tok, '=');
        char* addr  = eq ? eq + 1 : tok;
        char* colon = strrchr(addr, ':');
        if (!colon)
            return -1;
        *colon = '\0';
        xce_candidate_t* cand = &cfg->candidates[cfg->candidates_count];
        if (eq) {
            *eq = '\0';
            strncpy(cand->name, tok, XCE_MAXNAME_LEN - 1);
            cand->name[XCE_MAXNAME_LEN - 1] = '\0';
        } else {
            cand->name[0] = '\0';
        }
        strncpy(cand->hostname, addr, XCE_MAXHOSTNAME_LEN - 1);
        cand->hostname[XCE_MAXHOSTNAME_LEN - 1] = '\0';
        cand->port = atoi(colon + 1);
        cfg->candidates_count++;
        tok = strtok_r(NULL, ",", &saveptr);
    }
    return cfg->candidates_count > 0 ? 0 : -1;
}

/* ── Test seam ─────────────────────────────────────────────────── */

static xce_connect_fn_t s_connect_fn = NULL;

void xce_cluster_set_connect_fn(xce_connect_fn_t fn)
{
    s_connect_fn = fn;
}

/* do_connect: calls the mock when set, otherwise returns NULL (stub for
 * the real SSL connect via netio — to be wired in a later task). */
static void* do_connect(const char* host, int port)
{
    if (s_connect_fn)
        return s_connect_fn(host, port);
    /* Production path: real SSL connect via netio — stub for now */
    (void)host;
    (void)port;
    return NULL;
}

/* ── Validation ────────────────────────────────────────────────── */

int xce_cluster_config_validate(const xce_cluster_config_t* cfg)
{
    if (!cfg)
        return -EINVAL;
    if (cfg->candidates_count < 1 || cfg->candidates_count > MAX_HA_CANDIDATES)
        return -EINVAL;
    for (int i = 0; i < cfg->candidates_count; i++) {
        if (cfg->candidates[i].hostname[0] == '\0')
            return -EINVAL;
        if (cfg->candidates[i].port <= 0)
            return -EINVAL;
    }
    if (cfg->identity[0] == '\0')
        return -EINVAL;
    return 0;
}

/* ── Config conversion ─────────────────────────────────────────── */

/* Mirrors xce_test's inline masters-array population at
 * src/controller/testsuite/test.c:360-372, generalized into a reusable
 * function. out->masters[i] are pointers into cfg->candidates[i].hostname,
 * not copies -- same non-owning-storage contract xce_test already relies on
 * (see that file's comment on master_cfg's static lifetime), so the caller
 * must keep cfg alive at least as long as out is in use. */
int xce_cluster_config_to_client_config(const xce_cluster_config_t* cfg,
                                        xce_client_config_t* out)
{
    if (!cfg || !out)
        return -EINVAL;
    if (cfg->candidates_count < 1)
        return -EINVAL;

    int count = cfg->candidates_count;
    if (count > N_MAX_MASTERS) {
        fprintf(stderr,
                "WARN: HA cluster has %d candidates; using first %d "
                "(N_MAX_MASTERS)\n",
                count, N_MAX_MASTERS);
        count = N_MAX_MASTERS;
    }

    out->nmasters = count;
    for (int i = 0; i < count; i++) {
        out->masters[i]      = (char*)cfg->candidates[i].hostname;
        out->master_ports[i] = cfg->candidates[i].port;
    }

    return 0;
}

/* ── Connect with candidate fallback ───────────────────────────── */

void* xce_connect_cluster(xce_cluster_config_t* cfg)
{
    if (xce_cluster_config_validate(cfg) != 0)
        return NULL;

    for (int i = 0; i < cfg->candidates_count; i++) {
        void* ep = do_connect(cfg->candidates[i].hostname,
                              cfg->candidates[i].port);
        if (ep)
            return ep;
    }
    return NULL;
}

/* ── Reconnect state machine ───────────────────────────────────── */

int xce_reconnect_init(xce_reconnect_ctx_t* rctx, xce_cluster_config_t* cfg)
{
    if (!rctx || !cfg)
        return -EINVAL;
    memset(rctx, 0, sizeof(*rctx));
    rctx->config     = cfg;
    rctx->state      = XCE_RECONNECT_IDLE;
    rctx->backoff_ms = XCE_RECONNECT_INITIAL_BACKOFF_MS;
    return 0;
}

int xce_reconnect_step(xce_reconnect_ctx_t* rctx)
{
    if (!rctx)
        return -EINVAL;

    switch (rctx->state) {
    case XCE_RECONNECT_IDLE:
        rctx->state = XCE_RECONNECT_CONNECTING;
        return 0; /* next step does the actual connect */

    case XCE_RECONNECT_CONNECTING: {
        void* ep = xce_connect_cluster(rctx->config);
        if (ep) {
            rctx->endpoint   = ep;
            rctx->state      = XCE_RECONNECT_CONNECTED;
            rctx->attempt    = 0;
            rctx->backoff_ms = XCE_RECONNECT_INITIAL_BACKOFF_MS;
        } else {
            rctx->attempt++;
            rctx->state = XCE_RECONNECT_BACKOFF;
        }
        return 0;
    }

    case XCE_RECONNECT_BACKOFF: {
        /* Skip sleep in test context (mock connect set) to keep tests fast */
        if (!s_connect_fn) {
            struct timespec ts = {
                .tv_sec  = rctx->backoff_ms / 1000,
                .tv_nsec = (long)(rctx->backoff_ms % 1000) * 1000000L,
            };
            nanosleep(&ts, NULL);
        }
        /* Exponential backoff with cap */
        rctx->backoff_ms *= XCE_RECONNECT_BACKOFF_MULTIPLIER;
        if (rctx->backoff_ms > XCE_RECONNECT_MAX_BACKOFF_MS)
            rctx->backoff_ms = XCE_RECONNECT_MAX_BACKOFF_MS;
        rctx->state = XCE_RECONNECT_CONNECTING;
        return 0;
    }

    case XCE_RECONNECT_CONNECTED:
    case XCE_RECONNECT_FAILED:
        return -1;
    }
    return 0;
}

void xce_reconnect_fini(xce_reconnect_ctx_t* rctx)
{
    if (rctx)
        memset(rctx, 0, sizeof(*rctx));
}