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 "repl.h"
#include "mylock.h"

/* ============================================================
 * 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);

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

    mylock_unlock(&r->lock);

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

/* ============================================================
 * Replication send — stub in P2-1; wired in P2-4
 * ============================================================ */

int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
                   const void *payload, uint32_t payload_len)
{
    (void)r;
    (void)payload;
    (void)payload_len;

    /* WAL_CHECKPOINT entries must never be replicated */
    assert(hdr != NULL);
    assert(hdr->type != WAL_CHECKPOINT);

    /* Full implementation in P2-4 */
    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;
}

/* ============================================================
 * 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;
}