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>

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;
}

/* ── P1-3a: promotion / demotion / SIGUSR1 ─────────────────── */

/*
 * Async-signal-safe: the handler only performs an atomic store on an
 * int-width type, which is async-signal-safe on all POSIX platforms
 * (POSIX.1-2017 §2.4.3).  The pointer is set once by ha_install_sigusr1()
 * before signals are enabled; no locking is 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)
{
    /* Step 1: Increment election term — monotonically increasing fencing token */
    atomic_fetch_add_explicit(&ha->term, 1, memory_order_acq_rel);

    /* Step 2: Set role to PRIMARY */
    atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY, memory_order_release);

    /* Step 3: Reset last_received_crc chain (COH-5 — new primary starts fresh chain) */
    ha->last_received_crc = WAL_CRC_NONE;

    /* Steps 4+: hook re-registration, standby heartbeat, cluster_start() —
     * deferred to ha_promote_to_primary(cluster_t*) in P1-5. */
}

void ha_do_demote(ha_state_t *ha)
{
    /* Step 1: Set role to STANDBY — vcpu threads will reject new primary ops */
    atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY, memory_order_release);

    /* Step 2: Reset hook idempotency guard — next promote will re-register */
    ha->hooks_registered = 0;

    /* Step 3: hook disable done in ha_demote_to_standby(cluster_t*) in P1-5
     * because it needs the cluster's hooksys_t. */
}

/* ── 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 — full promotion in P1-3a.
         * For now: set failover_requested so watchdog thread can act. */
        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);

        /* Handle manual failover request (SIGUSR1 path — P1-3a) */
        if (atomic_load_explicit(&ha->failover_requested, memory_order_acquire)) {
            atomic_store_explicit(&ha->failover_requested, 0, memory_order_release);
            /* Full ha_promote_to_primary(c) call in P1-3a when cluster_t is available */
        }
    }
    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);
}

/*
 * The three cluster_t-dependent stubs below cannot dereference c->ha until
 * P1-5 adds `ha_state_t ha` to cluster_t.  They compile as no-ops now and
 * will be fleshed out in P1-3a after P1-5 wires the field in.
 */

int am_i_primary(cluster_t *c)
{
    (void)c;
    /* P1-3 adds real election probing.
     * Stub: always report not-primary until cluster wiring is in place. */
    return 0;
}

int ha_promote_to_primary(cluster_t *c)
{
    (void)c;
    /* Full sequence in P1-3a (WAL flush, hook calls, peer notify).
     * Stub: no-op — role management moved to P1-3a. */
    return 0;
}

int ha_demote_to_standby(cluster_t *c)
{
    (void)c;
    /* Full sequence in P1-3a.
     * Stub: no-op — role management moved to P1-3a. */
    return 0;
}