myxce/src/controller/repl.c

/*
 * repl.c — P2-1: Replicator lifecycle, slot management, and WAL trim guard.
 *
 * mylock_t is a plain pthread_mutex_t alias (see mylock.h).
 * API: mylock_init / mylock_fini / mylock_lock / mylock_unlock
 */
#include <assert.h>
#include <errno.h>
#include <string.h>

#include "controller.capnp.h"  /* OPCode_walAck, OPCode_upgradeConfigAck */
#include "repl.h"
#include "mylock.h"
#include "utils.h"
#include "request.h"   /* xce_wal_batch_req_raw */
#include "typemap.h"   /* wal_catchup_entry_t */

#ifdef XCE_TESTING
/* Test-injectable send hook: replaces xce_wal_batch_req_raw in unit tests. */
int (*repl_broadcast_send_hook)(void *ep,
                                wal_catchup_entry_t **entries,
                                int n_entries) = NULL;
#endif

/* ============================================================
 * Lifecycle
 * ============================================================ */

int repl_init(replicator_t *r)
{
    if (r == NULL) {
        return -EINVAL;
    }

    memset(r, 0, sizeof(*r));

    int rc = mylock_init(&r->lock);
    if (rc != 0) {
        return rc;
    }

    rc = pthread_mutex_init(&r->ack_lock, NULL);
    if (rc != 0) {
        mylock_fini(&r->lock);
        return -rc;
    }

    rc = pthread_cond_init(&r->ack_cond, NULL);
    if (rc != 0) {
        pthread_mutex_destroy(&r->ack_lock);
        mylock_fini(&r->lock);
        return -rc;
    }

    return 0;
}

void repl_fini(replicator_t *r)
{
    if (r == NULL) {
        return;
    }

    pthread_cond_destroy(&r->ack_cond);
    pthread_mutex_destroy(&r->ack_lock);
    mylock_fini(&r->lock);
}

/* ============================================================
 * Standby slot management
 * ============================================================ */

int repl_add_standby(replicator_t *r, repl_ep_t ep, const char *host, uint32_t port)
{
    if (r == NULL || ep == NULL || host == NULL) {
        return -EINVAL;
    }

    mylock_lock(&r->lock);

    /* Idempotency guard: if ep is already registered, return its slot index.
     * Req 1.3: duplicate repl_add_standby calls are treated as no-ops. */
    for (int i = 0; i < MAX_STANDBYS; i++) {
        if (r->standbys[i].ep == ep) {
            mylock_unlock(&r->lock);
            return i;
        }
    }

    if (r->nstandbys >= MAX_STANDBYS) {
        mylock_unlock(&r->lock);
        return -ENOSPC;
    }

    /* Find first empty slot */
    int idx = -1;
    for (int i = 0; i < MAX_STANDBYS; i++) {
        if (r->standbys[i].ep == NULL) {
            idx = i;
            break;
        }
    }

    if (idx < 0) {
        /* Should not happen given nstandbys guard above, but be safe */
        mylock_unlock(&r->lock);
        return -ENOSPC;
    }

    standby_conn_t *s = &r->standbys[idx];
    memset(s, 0, sizeof(*s));
    s->ep   = ep;
    s->port = port;
    strncpy(s->host, host, XCE_MAXNAME_LEN - 1);
    s->host[XCE_MAXNAME_LEN - 1] = '\0';
    atomic_store(&s->acked_lsn, 0);
    atomic_store(&s->acked_config_lsn, 0);
    atomic_store(&s->in_catch_up, 0);

    r->nstandbys++;

    mylock_unlock(&r->lock);
    return idx;
}

void repl_remove_standby(replicator_t *r, repl_ep_t ep)
{
    if (r == NULL || ep == NULL) {
        return;
    }

    mylock_lock(&r->lock);

    for (int i = 0; i < MAX_STANDBYS; i++) {
        standby_conn_t *s = &r->standbys[i];
        if (s->ep == ep) {
            s->last_disconnected_lsn = atomic_load(&s->acked_lsn);
            s->last_disconnect_time  = time(NULL);
            s->ep = NULL;
            r->nstandbys--;
            break;
        }
    }

    mylock_unlock(&r->lock);
}

void repl_expire_disconnect_slots(replicator_t *r, time_t now_s, uint32_t snap_interval_s)
{
    if (r == NULL) {
        return;
    }

    mylock_lock(&r->lock);

    for (int i = 0; i < MAX_STANDBYS; i++) {
        standby_conn_t *s = &r->standbys[i];
        /* Only process disconnected (ep==NULL) slots that still hold a guard LSN */
        if (s->ep == NULL && s->last_disconnected_lsn != 0) {
            /* snap_interval_s==0 means expire immediately regardless of wall clock */
            int expired = (snap_interval_s == 0) ||
                          (now_s >= s->last_disconnect_time + (time_t)snap_interval_s);
            if (expired) {
                memset(s, 0, sizeof(*s));
            }
        }
    }

    mylock_unlock(&r->lock);
}

/* ============================================================
 * ACK handling — call from io_worker read callback ONLY
 * ============================================================ */

void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn)
{
    if (r == NULL || ep == NULL) {
        return;
    }

    mylock_lock(&r->lock);

    for (int i = 0; i < MAX_STANDBYS; i++) {
        standby_conn_t *s = &r->standbys[i];
        if (s->ep == ep) {
            atomic_store(&s->acked_lsn, acked_lsn);
            if (s->pending_ack_count > 0)
                s->pending_ack_count--;
            break;
        }
    }

    mylock_unlock(&r->lock);

    /* Signal any waiter (repl_wait_ack / repl_wait_majority_ack) */
    pthread_mutex_lock(&r->ack_lock);
    pthread_cond_signal(&r->ack_cond);
    pthread_mutex_unlock(&r->ack_lock);
}

/* ============================================================
 * Replication send (task 7.2)
 *
 * Iterates active standbys; skips empty slots and standbys in catch-up.
 * Evicts standbys whose pending_ack_count exceeds MAX_PENDING_ACKS by marking
 * them in_catch_up=1 (inlined to avoid re-entrant lock acquisition).
 * ============================================================ */

int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
                   const void *payload, uint32_t payload_len)
{
    /* WAL_CHECKPOINT entries must never be replicated */
    assert(hdr != NULL);
    assert(hdr->type != WAL_CHECKPOINT);

    if (r == NULL)
        return -EINVAL;

    mylock_lock(&r->lock);

    if (r->nstandbys == 0) {
        WARN("repl_broadcast: no standbys registered — WAL entry lsn=%lu type=%u"
             " not replicated",
             (unsigned long)hdr->lsn, hdr->type);
        mylock_unlock(&r->lock);
        return 0;
    }

    for (int i = 0; i < MAX_STANDBYS; i++) {
        standby_conn_t *s = &r->standbys[i];

        if (s->ep == NULL)
            continue;

        if (atomic_load_explicit(&s->in_catch_up, memory_order_acquire))
            continue;

        if (s->pending_ack_count > MAX_PENDING_ACKS) {
            /* Inline repl_start_catchup to avoid re-entrant lock acquisition */
            s->join_applied_lsn = atomic_load_explicit(&s->acked_lsn,
                                                        memory_order_acquire);
            atomic_store_explicit(&s->in_catch_up, 1, memory_order_release);
            WARN("repl_broadcast: standby %s:%u evicted (pending_ack_count=%u > %u)"
                 " — triggering catch-up reconnect",
                 s->host, s->port, s->pending_ack_count, MAX_PENDING_ACKS);
            continue;
        }

        /* Build a single-entry batch and enqueue it for this standby. */
        wal_catchup_entry_t entry;
        memset(&entry, 0, sizeof(entry));
        entry.lsn             = hdr->lsn;
        entry.entry_type      = (uint8_t)hdr->type;
        entry.raw_payload     = (uint8_t *)(uintptr_t)payload;
        entry.raw_payload_len = payload_len;

        wal_catchup_entry_t *ep_arr[1] = { &entry };

        int rc;
#ifdef XCE_TESTING
        if (repl_broadcast_send_hook)
            rc = repl_broadcast_send_hook(s->ep, ep_arr, 1);
        else
            rc = xce_wal_batch_req_raw(s->ep, ep_arr, 1);
#else
        rc = xce_wal_batch_req_raw(s->ep, ep_arr, 1);
#endif
        if (rc != 0) {
            WARN("repl_broadcast: send to standby %s:%u failed (rc=%d) — continuing",
                 s->host, s->port, rc);
            continue;
        }

        TRACE("repl_broadcast: sent lsn=%lu type=%u to %s:%u",
              (unsigned long)hdr->lsn, hdr->type, s->host, s->port);
        s->pending_ack_count++;
    }

    mylock_unlock(&r->lock);
    return 0;
}

/* ============================================================
 * Sync replication — stub in P2-1; wired in P2-3
 * ============================================================ */

int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms)
{
    if (!r) return -EINVAL;

    /* Check whether any non-catching-up standby exists */
    mylock_lock(&r->lock);
    int has_standby = 0;
    for (int i = 0; i < MAX_STANDBYS; i++) {
        if (r->standbys[i].ep != NULL &&
            !atomic_load_explicit(&r->standbys[i].in_catch_up,
                                  memory_order_acquire)) {
            has_standby = 1;
            break;
        }
    }
    mylock_unlock(&r->lock);

    if (!has_standby) return -ENODEV;

    /* Compute absolute deadline for pthread_cond_timedwait (CLOCK_REALTIME) */
    struct timespec deadline;
    clock_gettime(CLOCK_REALTIME, &deadline);
    deadline.tv_sec  += (long)(timeout_ms / 1000);
    deadline.tv_nsec += (long)(timeout_ms % 1000) * 1000000L;
    if (deadline.tv_nsec >= 1000000000L) {
        deadline.tv_sec++;
        deadline.tv_nsec -= 1000000000L;
    }

    pthread_mutex_lock(&r->ack_lock);
    int rc = -ETIMEDOUT;
    while (1) {
        /* Recheck every standby after each wakeup (spurious-wakeup guard) */
        mylock_lock(&r->lock);
        for (int i = 0; i < MAX_STANDBYS; i++) {
            standby_conn_t *s = &r->standbys[i];
            if (s->ep == NULL) continue;
            if (atomic_load_explicit(&s->in_catch_up, memory_order_acquire))
                continue;
            if (atomic_load_explicit(&s->acked_lsn, memory_order_acquire)
                    >= target_lsn) {
                rc = 0;
                break;
            }
        }
        mylock_unlock(&r->lock);
        if (rc == 0) break;

        int wait_rc = pthread_cond_timedwait(&r->ack_cond, &r->ack_lock,
                                             &deadline);
        if (wait_rc == ETIMEDOUT) {
            rc = -ETIMEDOUT;
            break;
        }
    }
    pthread_mutex_unlock(&r->ack_lock);
    return rc;
}

/* ============================================================
 * Config ACK handling — call from io_worker read callback ONLY (task 8.1)
 * ============================================================ */

void repl_on_config_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_config_lsn)
{
    if (r == NULL || ep == NULL)
        return;

    mylock_lock(&r->lock);
    for (int i = 0; i < MAX_STANDBYS; i++) {
        standby_conn_t *s = &r->standbys[i];
        if (s->ep == ep) {
            atomic_store_explicit(&s->acked_config_lsn, acked_config_lsn,
                                  memory_order_release);
            break;
        }
    }
    mylock_unlock(&r->lock);

    pthread_mutex_lock(&r->ack_lock);
    pthread_cond_signal(&r->ack_cond);
    pthread_mutex_unlock(&r->ack_lock);
}

/* ============================================================
 * Majority-quorum ACK wait (task 7.3)
 *
 * Quorum formula: quorum = (n_active + 2) / 2  [ceiling of (n_active+1)/2]
 *   n_active = count of active (non-catch-up) standbys.
 *
 * walAck:    acked_total = 1 (self) + standbys with acked_lsn > 0.
 *            Primary counts itself — it persists WAL before broadcasting.
 * configAck: acked_total = standbys with acked_config_lsn > 0 (no self-count).
 *            Config must be confirmed by standbys; primary does not self-count.
 *
 * Returns 0 on quorum, -ETIMEDOUT on timeout.
 * ============================================================ */

int repl_wait_majority_ack(replicator_t *r, uint64_t entry_lsn,
                           uint32_t timeout_ms, uint8_t ack_op)
{
    /* entry_lsn is accepted but unused by this legacy counting body
     * (ha-commit-engine task 5.1 signature change): task 5.2 delegates the
     * walAck branch to commit_submit(ce, entry_lsn, COMMIT_MODE_WAIT, ...)
     * when the engine is wired. Behavior here stays byte-identical. */
    (void)entry_lsn;

    if (r == NULL)
        return -EINVAL;

    int is_wal_ack = (ack_op == (uint8_t)OPCode_walAck);

    struct timespec deadline;
    clock_gettime(CLOCK_REALTIME, &deadline);
    deadline.tv_sec  += (long)(timeout_ms / 1000);
    deadline.tv_nsec += (long)(timeout_ms % 1000) * 1000000L;
    if (deadline.tv_nsec >= 1000000000L) {
        deadline.tv_sec++;
        deadline.tv_nsec -= 1000000000L;
    }

    pthread_mutex_lock(&r->ack_lock);
    int rc = -ETIMEDOUT;

    while (1) {
        mylock_lock(&r->lock);

        int n_active    = 0;
        int acked_count = 0;

        for (int i = 0; i < MAX_STANDBYS; i++) {
            standby_conn_t *s = &r->standbys[i];
            if (s->ep == NULL)
                continue;
            if (atomic_load_explicit(&s->in_catch_up, memory_order_acquire))
                continue;
            n_active++;
            if (is_wal_ack) {
                if (atomic_load_explicit(&s->acked_lsn, memory_order_acquire) > 0)
                    acked_count++;
            } else {
                if (atomic_load_explicit(&s->acked_config_lsn,
                                         memory_order_acquire) > 0)
                    acked_count++;
            }
        }

        mylock_unlock(&r->lock);

        /* Single-node shortcircuit: no active standbys → local write is quorum.
         * Applies to both walAck (self=1, quorum=1) and configAck (Req 8.5). */
        if (n_active == 0) {
            rc = 0;
            break;
        }

        int quorum      = (n_active + 2) / 2;
        int acked_total = is_wal_ack ? (1 + acked_count) : acked_count;

        if (acked_total >= quorum) {
            rc = 0;
            break;
        }

        int wait_rc = pthread_cond_timedwait(&r->ack_cond, &r->ack_lock,
                                             &deadline);
        if (wait_rc == ETIMEDOUT) {
            rc = -ETIMEDOUT;
            break;
        }
    }

    pthread_mutex_unlock(&r->ack_lock);

    if (rc == 0)
        INFO("repl_wait_majority_ack: quorum reached (ack_op=%u)", ack_op);
    else
        WARN("repl_wait_majority_ack: timeout waiting for majority quorum"
             " (ack_op=%u, timeout_ms=%u)", ack_op, timeout_ms);

    return rc;
}

/* ============================================================
 * WAL trim guard
 * ============================================================ */

uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn,
                        uint32_t snap_interval_s)
{
    (void)snap_interval_s;

    if (r == NULL) {
        return checkpoint_lsn;
    }

    uint64_t min_lsn = checkpoint_lsn;

    mylock_lock(&r->lock);

    for (int i = 0; i < MAX_STANDBYS; i++) {
        const standby_conn_t *s = &r->standbys[i];

        if (s->ep != NULL) {
            /* Active standby: guard at its last acked LSN */
            uint64_t acked = atomic_load(&s->acked_lsn);
            if (acked < min_lsn) {
                min_lsn = acked;
            }
        } else if (s->last_disconnected_lsn != 0) {
            /* Recently disconnected slot still holding a guard */
            if (s->last_disconnected_lsn < min_lsn) {
                min_lsn = s->last_disconnected_lsn;
            }
        }
    }

    mylock_unlock(&r->lock);

    return min_lsn;
}

/* ============================================================
 * Catch-up state management (P2-5)
 * ============================================================ */

void repl_start_catchup(replicator_t *r, repl_ep_t ep, uint64_t join_applied_lsn)
{
    mylock_lock(&r->lock);
    for (int i = 0; i < MAX_STANDBYS; i++) {
        if (r->standbys[i].ep == ep) {
            r->standbys[i].join_applied_lsn = join_applied_lsn;
            atomic_store_explicit(&r->standbys[i].in_catch_up, 1,
                                  memory_order_release);
            break;
        }
    }
    mylock_unlock(&r->lock);
}

void repl_complete_catchup(replicator_t *r, repl_ep_t ep)
{
    mylock_lock(&r->lock);
    for (int i = 0; i < MAX_STANDBYS; i++) {
        if (r->standbys[i].ep == ep) {
            atomic_store_explicit(&r->standbys[i].in_catch_up, 0,
                                  memory_order_release);
            break;
        }
    }
    mylock_unlock(&r->lock);

    /* Broadcast so any blocked repl_wait_ack() re-checks the ACK condition */
    pthread_mutex_lock(&r->ack_lock);
    pthread_cond_broadcast(&r->ack_cond);
    pthread_mutex_unlock(&r->ack_lock);
}

int repl_needs_snapshot(replicator_t *r, uint64_t standby_lsn,
                        uint64_t oldest_wal_lsn)
{
    (void)r;
    /* No WAL at all: empty standby can start from LSN 0 without a snapshot */
    if (oldest_wal_lsn == 0) return 0;
    /* Standby LSN predates oldest surviving WAL segment — full snapshot required */
    return (standby_lsn < oldest_wal_lsn) ? 1 : 0;
}

/* ============================================================
 * Write-admission quorum accounting (ha-commit-engine task 4)
 * ============================================================ */

/* repl_active_standby_count — registry-truth count of standbys able to
 * receive new log entries: slots with ep != NULL && !in_catch_up, evaluated
 * under r->lock. NULL registry counts zero. Consumed by ha_has_write_quorum
 * (ha.c) for write admission. Takes ONLY r->lock — never the commit-engine
 * mutex (lock-ordering rule: gate path holds no locks when calling in).
 * (ha-commit-engine Req 3.1, 3.2, 3.3) */
/* repl_set_commit_engine — INERT stub (ha-commit-engine task 5.1).
 * Stores nothing, so the engine is never wired and the task-5.1 RED tests
 * fail on behavior (ACK and membership events never reach the engine),
 * not on linking. Task 5.2 adds the replicator_t.ce field, stores the
 * pointer here, and feeds commit_on_ack / commit_on_membership_change
 * from repl_on_ack and the membership mutators. */
void repl_set_commit_engine(replicator_t *r, struct commit_engine *ce)
{
    (void)r;
    (void)ce;
}

int repl_active_standby_count(replicator_t *r)
{
    int i;
    int count = 0;

    if (!r)
        return 0;

    mylock_lock(&r->lock);
    for (i = 0; i < MAX_STANDBYS; i++) {
        if (r->standbys[i].ep != NULL &&
            !atomic_load_explicit(&r->standbys[i].in_catch_up,
                                  memory_order_acquire))
            count++;
    }
    mylock_unlock(&r->lock);

    return count;
}