myxce/src/controller/commit.c

/*
 * commit.c — Commit engine core (ha-commit-engine task 1.2).
 *
 * Single authority for per-entry commit evaluation and committed_lsn
 * advancement (Raft §5.3, umbrella-verbatim commit rule):
 *
 *   entry N is committed when
 *     count(self + active standbys with acked_lsn >= N) > n_peers / 2
 *
 * Evaluated as a watermark median: active (connected, non-catch-up)
 * standby watermarks are scanned into a scratch array, sorted descending,
 * and the candidate is arr[n_peers/2 - 1] — the highest LSN covered by
 * self plus a majority of peers. The candidate is clamped to the highest
 * locally appended LSN and stored with monotonic max() semantics.
 *
 * commit_on_ack is called from the iomgr read path and never blocks
 * beyond the engine mutex (plus the registry slot mutex, which the same
 * iomgr path already takes in repl_on_ack): no I/O, no condvar waits.
 *
 * COMMIT_MODE_WAIT (task 2.2) blocks the caller until its per-LSN
 * predicate holds — own entry committed AND the own-term gate (when set)
 * reached — or the deadline expires (-ETIMEDOUT). Admission control caps
 * concurrent blocking waits at max(1, n_vcpus - COMMIT_WAIT_RESERVED);
 * cap exhaustion is the ONLY cause of -XCE_ERR_NOT_READY (an unmet gate
 * blocks within the timeout instead — CORRECTED contract, design
 * §Commit Engine). n_peers <= 1 self-quorum advances committed_lsn
 * through the single advancement path and completes immediately.
 *
 * Requirements: 1.1, 1.2, 2.1, 2.2, 2.3, 4.3, 4.4, 5.1, 6.3, 7.1, 7.2, 7.3
 */
#include <errno.h>
#include <stdatomic.h>
#include <string.h>
#include <time.h>

#include "xce/errnox.h"

#include "commit.h"
#include "mylock.h"
#include "utils.h"

#ifdef XCE_TESTING
#include <assert.h>
#endif

/* ============================================================
 * Internal helpers
 * ============================================================ */

/* Highest locally appended WAL LSN. persist invariant:
 * committed_lsn <= current_lsn - 1 (current_lsn is the NEXT LSN). */
static uint64_t highest_appended_lsn(const persist_t* p)
{
    return (p->current_lsn > 0) ? p->current_lsn - 1 : 0;
}

/* Descending insertion sort — n is bounded by MAX_STANDBYS. */
static void sort_desc(uint64_t* arr, int n)
{
    int      i, j;
    uint64_t v;

    for (i = 1; i < n; i++) {
        v = arr[i];
        for (j = i - 1; j >= 0 && arr[j] < v; j--) {
            arr[j + 1] = arr[j];
        }
        arr[j + 1] = v;
    }
}

/*
 * Compute the highest LSN covered by self + a majority of peers.
 * Caller holds ce->lock.
 *
 * Catch-up standbys are excluded from the evaluation count
 * (umbrella-locked conservative rule); absent peers count as watermark 0.
 */
static uint64_t majority_candidate_locked(commit_engine_t* ce)
{
    replicator_t* r;
    uint64_t      cand;
    uint64_t      cap = highest_appended_lsn(ce->p);
    int           n   = 0;
    int           need, k, i;

    /* Standalone / single-member: self constitutes the majority (Req 6.3). */
    if (ce->n_peers <= 1) {
        return cap;
    }

    /* Bounded scan of active standby watermarks into the scratch array. */
    r = &ce->p->repl;
    mylock_lock(&r->lock);
    for (i = 0; i < MAX_STANDBYS && n < MAX_STANDBYS; i++) {
        standby_conn_t* s = &r->standbys[i];

        if (s->ep == NULL || atomic_load(&s->in_catch_up) != 0) {
            continue; /* empty slot or catch-up: not countable toward majority */
        }
        ce->scratch[n++] = atomic_load(&s->acked_lsn);
    }
    mylock_unlock(&r->lock);

    /* Zero-fill up to the full peer count so missing peers count as 0. */
    need = ce->n_peers - 1;
    if (need > MAX_STANDBYS) {
        need = MAX_STANDBYS;
    }
    while (n < need) {
        ce->scratch[n++] = 0;
    }

    /* Self always covers every appended entry, so a majority needs
     * n_peers/2 standbys at-or-above N; sorted descending, the highest
     * such N is the watermark at index n_peers/2 - 1. */
    k = ce->n_peers / 2 - 1;
    if (k >= n) {
        return 0; /* peer count not representable — no majority possible */
    }
    sort_desc(ce->scratch, n);
    cand = ce->scratch[k];

    /* Contiguous and bounded: never beyond the appended log (Req 2.3). */
    return (cand < cap) ? cand : cap;
}

/*
 * Re-evaluate majority and advance committed_lsn (monotonic max() store).
 * Caller holds ce->lock. Broadcasts the condvar on advancement so future
 * WAIT-mode blockers (task 2.2) re-check their per-LSN predicates.
 */
static void commit_advance_locked(commit_engine_t* ce)
{
    uint64_t cand = majority_candidate_locked(ce);
    uint64_t cur  = atomic_load(&ce->p->committed_lsn);

    if (cand <= cur) {
        return; /* monotonic: committed_lsn never regresses (Req 2.2) */
    }

#ifdef XCE_TESTING
    assert(cand > cur);                              /* monotonic advance only */
    assert(cand <= highest_appended_lsn(ce->p));     /* never beyond the log */
#endif

    atomic_store(&ce->p->committed_lsn, cand);
    TRACE("commit: committed_lsn %lu -> %lu (n_peers=%d)", cur, cand,
          ce->n_peers);
    pthread_cond_broadcast(&ce->cond);
}

/* Admission cap for concurrent blocking waits, expressed only relative to
 * the n_vcpus init argument: keep COMMIT_WAIT_RESERVED vtasks free of
 * blocking commit waits, floored at 1 so one wait is always admittable
 * (Req 7.1, 7.2). */
static int wait_cap(const commit_engine_t* ce)
{
    int cap = ce->n_vcpus - COMMIT_WAIT_RESERVED;

    return (cap < 1) ? 1 : cap;
}

/* WAIT completion predicate (Req 1.1): own entry committed AND the
 * own-term gate (when set) reached. Caller holds ce->lock. */
static int wait_done_locked(const commit_engine_t* ce, uint64_t lsn)
{
    uint64_t c = atomic_load(&ce->p->committed_lsn);

    return c >= lsn && (ce->writable_lsn == 0 || c >= ce->writable_lsn);
}

/* Absolute deadline now + timeout_ms on CLOCK_REALTIME — the condvar was
 * initialized with default attrs (task 1.2), so pthread_cond_timedwait
 * measures against CLOCK_REALTIME. */
static void wait_deadline(struct timespec* dl, uint32_t timeout_ms)
{
    clock_gettime(CLOCK_REALTIME, dl);
    dl->tv_sec += timeout_ms / 1000u;
    dl->tv_nsec += (long)(timeout_ms % 1000u) * 1000000L;
    if (dl->tv_nsec >= 1000000000L) {
        dl->tv_sec += 1;
        dl->tv_nsec -= 1000000000L;
    }
}

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

int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcpus)
{
    if (ce == NULL || p == NULL || n_peers < 0 || n_vcpus < 1) {
        return -EINVAL;
    }
    pthread_mutex_init(&ce->lock, NULL);
    pthread_cond_init(&ce->cond, NULL);
    ce->p              = p;
    ce->n_peers        = n_peers;
    ce->n_vcpus        = n_vcpus;
    ce->inflight_waits = 0;
    ce->writable_lsn   = 0; /* no own-term gate until set by promotion */
    memset(ce->scratch, 0, sizeof(ce->scratch));
    return 0;
}

void commit_engine_fini(commit_engine_t* ce)
{
    if (ce == NULL) {
        return;
    }
    pthread_cond_destroy(&ce->cond);
    pthread_mutex_destroy(&ce->lock);
    ce->p = NULL;
}

/* ============================================================
 * Entry registration
 * ============================================================ */

int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode,
                  uint32_t timeout_ms)
{
    struct timespec dl;
    uint64_t        cur, gate;
    int             rc = 0;
    int             cap, wrc;

    if (ce == NULL || ce->p == NULL || lsn == 0) {
        return -EINVAL;
    }

    if (mode == COMMIT_MODE_RETAIN) {
        /* Commit state is fully derived from the registry watermarks at
         * ACK time, so RETAIN registration has nothing to record per
         * entry: return immediately, never block; the entry commits later
         * via commit_on_ack (Req 2.1). RETAIN never honors the timeout. */
        return 0;
    }

    /* COMMIT_MODE_WAIT — blocking wait with admission control. The
     * deadline is absolute so spurious wakeups never extend the wait. */
    wait_deadline(&dl, timeout_ms);

    pthread_mutex_lock(&ce->lock);

    /* Admission check: cap exhaustion is the ONLY -XCE_ERR_NOT_READY
     * cause (Req 7.1, 7.2; CORRECTED contract). Retriable fast-reject —
     * never blocks. */
    cap = wait_cap(ce);
    if (ce->inflight_waits >= cap) {
        pthread_mutex_unlock(&ce->lock);
        WARN("commit: WAIT rejected, admission cap exhausted "
             "(cap=%d n_vcpus=%d lsn=%lu)",
             cap, ce->n_vcpus, lsn);
        return -XCE_ERR_NOT_READY;
    }

    ce->inflight_waits++;
#ifdef XCE_TESTING
    assert(ce->inflight_waits > 0 && ce->inflight_waits <= cap);
#endif

    /* Standalone / single-member: self constitutes the quorum (Req 6.3).
     * Advance through the single advancement path — never a second
     * committed_lsn store site — then complete via the same predicate. */
    if (ce->n_peers <= 1) {
        commit_advance_locked(ce);
    }

    while (!wait_done_locked(ce, lsn)) {
        wrc = pthread_cond_timedwait(&ce->cond, &ce->lock, &dl);
        if (wrc != 0) {
            /* Deadline expired (or hard condvar error): one last
             * predicate check closes the wakeup/timeout race. */
            if (!wait_done_locked(ce, lsn)) {
                rc = -ETIMEDOUT;
            }
            break;
        }
    }

    ce->inflight_waits--;
#ifdef XCE_TESTING
    assert(ce->inflight_waits >= 0);
#endif
    cur  = atomic_load(&ce->p->committed_lsn);
    gate = ce->writable_lsn;
    pthread_mutex_unlock(&ce->lock);

    if (rc == -ETIMEDOUT) {
        WARN("commit: WAIT deadline expired without commit "
             "(lsn=%lu committed_lsn=%lu writable_lsn=%lu timeout_ms=%u)",
             lsn, cur, gate, timeout_ms);
    }
    return rc;
}

/* ============================================================
 * ACK and membership events
 * ============================================================ */

void commit_on_ack(commit_engine_t* ce, int standby_idx, uint64_t acked_lsn)
{
    if (ce == NULL || ce->p == NULL || standby_idx < 0 ||
        standby_idx >= MAX_STANDBYS) {
        return;
    }

    TRACE("commit: ack standby_idx=%d acked_lsn=%lu", standby_idx, acked_lsn);

    /* The authoritative watermark is read from the registry during
     * evaluation (repl_on_ack stored it before this call); the args
     * identify the event for tracing only. */
    pthread_mutex_lock(&ce->lock);
    commit_advance_locked(ce);
    pthread_mutex_unlock(&ce->lock);
}

void commit_set_writable_lsn(commit_engine_t* ce, uint64_t noop_lsn)
{
    if (ce == NULL) {
        return;
    }
    pthread_mutex_lock(&ce->lock);
    ce->writable_lsn = noop_lsn;
    /* Gate change is part of the WAIT completion predicate: wake blockers
     * so they re-evaluate (no successful completion until
     * committed_lsn >= writable_lsn). */
    pthread_cond_broadcast(&ce->cond);
    pthread_mutex_unlock(&ce->lock);
}

void commit_on_membership_change(commit_engine_t* ce)
{
    if (ce == NULL || ce->p == NULL) {
        return;
    }
    pthread_mutex_lock(&ce->lock);
    /* Catch-up transitions / standby removal can newly satisfy a pending
     * majority: re-evaluate (monotonic, so shrink never regresses) and
     * wake all waiters to re-check their predicates. */
    commit_advance_locked(ce);
    pthread_cond_broadcast(&ce->cond);
    pthread_mutex_unlock(&ce->lock);
}