myxce/src/controller/ha.c

#define _GNU_SOURCE
#include "ha.h"

#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.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"

int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority)
{
    if (!ha || !cfg)
        return -EINVAL;

    if (cfg->heartbeat_interval_ms < 2000)
        return -EINVAL;
    if (cfg->heartbeat_timeout_ms < cfg->heartbeat_interval_ms)
        return -EINVAL;
    if (cfg->election_timeout_ms < 500)
        return -EINVAL;

    memset(ha, 0, sizeof(*ha));
    ha->config      = *cfg;
    ha->my_priority = my_priority;

    atomic_store_explicit(&ha->role,                  (int32_t)CTRL_STANDBY, memory_order_release);
    atomic_store_explicit(&ha->term,                  0ULL, memory_order_release);
    atomic_store_explicit(&ha->applied_lsn,           0ULL, memory_order_release);
    atomic_store_explicit(&ha->promotion_failed,      0,    memory_order_relaxed);
    atomic_store_explicit(&ha->promotion_retry_count, 0,    memory_order_relaxed);
    atomic_store_explicit(&ha->in_catch_up,           0,    memory_order_relaxed);
    atomic_store_explicit(&ha->failover_requested,    0,    memory_order_relaxed);
    atomic_store_explicit(&ha->watchdog_stop,         0,    memory_order_relaxed);

    pthread_mutex_init(&ha->lock, NULL);
    return 0;
}

/* ── Promotion / demotion / SIGUSR1 ────────────────────────── */

/* Async-signal-safe: atomic store on int-width type (POSIX.1-2017 §2.4.3).
 * Pointer set once before signals are enabled — no locking required. */
static ha_state_t *g_sigusr1_ha = NULL;

static void sigusr1_handler(int sig)
{
    (void)sig;
    if (g_sigusr1_ha)
        atomic_store_explicit(&g_sigusr1_ha->failover_requested, 1,
                              memory_order_release);
}

void ha_install_sigusr1(ha_state_t *ha)
{
    g_sigusr1_ha = ha;
    struct sigaction sa;
    memset(&sa, 0, sizeof(sa));
    sa.sa_handler = sigusr1_handler;
    sigemptyset(&sa.sa_mask);
    sa.sa_flags = SA_RESTART;
    sigaction(SIGUSR1, &sa, NULL);
}

void ha_do_promote(ha_state_t *ha)
{
    /* Monotonically increasing fencing token — standbys reject entries from lower terms */
    atomic_fetch_add_explicit(&ha->term, 1, memory_order_acq_rel);
    atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY, memory_order_release);
    /* COH-5: new primary starts a fresh CRC chain; standbys reset their chain on catch-up */
    ha->last_received_crc = WAL_CRC_NONE;
}

void ha_do_demote(ha_state_t *ha)
{
    atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY, memory_order_release);
    /* Reset so the next ha_promote_to_primary() re-registers hooks idempotently */
    ha->hooks_registered = 0;
    /* Hook disable requires hooksys_t — done by ha_demote_to_standby(cluster_t*) */
}

/* ── P3-6: Test-only failover shim ────────────────────────────── */

#ifdef XCE_TESTING
uint64_t ha_trigger_manual_failover_for_test(ha_state_t *ha)
{
    ha_do_demote(ha);
    ha_do_promote(ha);
    return atomic_load_explicit(&ha->term, memory_order_acquire);
}
#endif

/* ── Watchdog thread implementation ────────────────────────── */

void ha_check_heartbeat_nocluster(ha_state_t *ha)
{
    ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);

    /* Only standbys watch for primary heartbeat timeout */
    if (role != CTRL_STANDBY) return;

    /* If we have no primary endpoint, nothing to monitor */
    if (ha->primary_host[0] == '\0') return;

    struct timespec now;
    clock_gettime(CLOCK_MONOTONIC, &now);

    pthread_mutex_lock(&ha->lock);
    struct timespec last = ha->last_heartbeat;
    pthread_mutex_unlock(&ha->lock);

    uint64_t elapsed_ms =
        (uint64_t)(now.tv_sec  - last.tv_sec)  * 1000ULL +
        (uint64_t)(now.tv_nsec - last.tv_nsec) / 1000000ULL;

    if (elapsed_ms >= ha->config.heartbeat_timeout_ms) {
        /* Primary appears dead — set failover_requested; watchdog acts on it */
        atomic_store_explicit(&ha->failover_requested, 1, memory_order_release);
    }
}

static void *ha_watchdog_thread(void *arg)
{
    ha_state_t *ha = arg;

    while (!atomic_load_explicit(&ha->watchdog_stop, memory_order_acquire)) {
        long ms = (long)ha->config.heartbeat_interval_ms;
        struct timespec ts = {
            .tv_sec  = ms / 1000,
            .tv_nsec = (ms % 1000) * 1000000L,
        };
        nanosleep(&ts, NULL);

        /* Re-check stop flag after sleep */
        if (atomic_load_explicit(&ha->watchdog_stop, memory_order_acquire)) break;

        ha_check_heartbeat_nocluster(ha);

        if (atomic_load_explicit(&ha->failover_requested, memory_order_acquire)) {
            atomic_store_explicit(&ha->failover_requested, 0, memory_order_release);
            /* Promotion requires cluster_t; the watchdog only has ha_state_t.
             * Full ha_promote_to_primary(c) is called from the cluster's own
             * failover path which holds cluster_t context. */
        }
    }
    return NULL;
}

int ha_start_watchdog(ha_state_t *ha)
{
    atomic_store_explicit(&ha->watchdog_stop, 0, memory_order_release);
    int rc = pthread_create(&ha->watchdog_thread, NULL, ha_watchdog_thread, ha);
    if (rc != 0) return -rc;
    return 0;
}

void ha_stop_watchdog(ha_state_t *ha)
{
    atomic_store_explicit(&ha->watchdog_stop, 1, memory_order_release);
    if (ha->watchdog_thread != 0) {
        pthread_join(ha->watchdog_thread, NULL);
        ha->watchdog_thread = 0;
    }
}

void ha_fini(ha_state_t *ha)
{
    if (!ha)
        return;
    ha_stop_watchdog(ha);  /* join watchdog thread before destroying mutex */
    pthread_mutex_destroy(&ha->lock);
}

/* ── Redirect helpers used by join.c ───────────────────────── */

/*
 * ha_should_redirect: returns 1 when this standby knows the primary's
 * address and should redirect incoming join requests there.
 * Returns 0 for PRIMARY role, or for STANDBY with no known primary yet.
 */
int ha_should_redirect(const ha_state_t *ha)
{
    ctrl_role_t role =
        (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
    if (role == CTRL_PRIMARY)
        return 0;
    if (ha->primary_host[0] == '\0' || ha->primary_port == 0)
        return 0;
    return 1;
}

/*
 * ha_no_primary_known: returns 1 when this node is a STANDBY but does not
 * yet know the primary's address.  Callers should reply E_XCE_NOMASTER so
 * the client retries rather than treating the error as permanent.
 */
int ha_no_primary_known(const ha_state_t *ha)
{
    ctrl_role_t role =
        (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
    if (role == CTRL_PRIMARY)
        return 0;
    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).
 * Safe to call on a zero-init cluster_t (zeroed hooksys → xce_add_hook
 * returns -EINVAL harmlessly; hooks_registered is still set to 1).
 */
int ha_promote_to_primary(cluster_t *c)
{
    ha_do_promote(&c->ha);

    /* Register persistence hooks — idempotency guard prevents double-add */
    if (c->ha.hooks_registered == 0) {
        xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_JOB,
                     (hook_fn_t)persist_job_hook_fun, &c->persist);
        xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_QUEUE,
                     (hook_fn_t)persist_queue_hook_fun, &c->persist);
        xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER,
                     (hook_fn_t)persist_worker_hook_fun, &c->persist);
        c->ha.hooks_registered = 1;
    }
    return 0;
}

/*
 * ha_demote_to_standby: disable persistence hooks, then set role=STANDBY
 * and reset the hooks_registered idempotency guard (via ha_do_demote).
 */
int ha_demote_to_standby(cluster_t *c)
{
    /* Disable persistence hooks before changing role */
    if (c->ha.hooks_registered) {
        xce_disable_hook(&c->hooksys, XCE_CONTROLLER_HOOK_JOB,
                         (hook_fn_t)persist_job_hook_fun);
        xce_disable_hook(&c->hooksys, XCE_CONTROLLER_HOOK_QUEUE,
                         (hook_fn_t)persist_queue_hook_fun);
        xce_disable_hook(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER,
                         (hook_fn_t)persist_worker_hook_fun);
    }
    ha_do_demote(&c->ha);   /* sets role=STANDBY, resets hooks_registered=0 */
    return 0;
}