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;
    /* DURABILITY PRECONDITION (ha-commit-durability Req 1.2): the PRIMARY's
     * self-contribution to the commit majority is capped at its DURABLE
     * watermark (the highest LSN fdatasync'd locally), NOT the appended tip.
     * An entry that is only in the page cache (appended but not yet synced)
     * must not count self toward its majority — otherwise "committed" could be
     * durable on a single member and lost across failover (bug-661). The
     * standby side already persists-before-ACK (persist_wal_write_replicated),
     * so once self is durable-capped, committed_lsn ⇒ durable on a majority.
     * highest_appended_lsn is still used below as the never-beyond-the-log
     * bound (it can only exceed durable_lsn, so it never loosens this cap). */
    uint64_t      cap = persist_durable_lsn(ce->p);
    int           n   = 0;
    int           need, k, i;

    /* Standalone / single-member: self constitutes the majority (Req 6.3) —
     * but self still only counts up to what is durable locally, so a
     * single-member commit also cannot advance over a not-yet-synced entry. */
    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 */
    /* DURABLE-COMMIT INVARIANT (ha-commit-durability task 2.4, Req 1.1/1.2):
     * committed_lsn must never advance past what is durable locally. The
     * candidate is already durable-capped by majority_candidate_locked (self
     * caps at persist_durable_lsn), so this must hold — it guards against a
     * future change that reintroduces an ACK-without-durability advance. */
    assert(cand <= persist_durable_lsn(ce->p));
#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;
    }
}

/* Durable-advance wake (ha-commit-durability task 2.2): invoked by the persist
 * group-commit flusher AFTER it advances durable_lsn and AFTER it releases
 * p->lock. Now that the commit majority caps self at durable_lsn, an advance of
 * durable_lsn can newly satisfy a pending entry's majority even when no standby
 * ACK arrived — re-evaluate and wake blocked COMMIT_MODE_WAIT callers so they
 * complete promptly instead of stalling until their timeout (the lost-wakeup
 * window: all standby ACKs landed before the local fdatasync finished, so
 * commit_on_ack already fired and nothing else would re-check).
 *
 * LOCK ORDERING: this runs on the flusher thread with NO persist lock held; it
 * takes only ce->lock. The commit side never takes p->lock while holding
 * ce->lock, so the only edge is p->lock → ce->lock (cycle-free, deadlock-free).
 * ctx is the commit_engine_t registered in commit_engine_init. */
static void commit_on_durable_advance(void* ctx)
{
    commit_engine_t* ce = ctx;

    if (ce == NULL || ce->p == NULL) {
        return;
    }
    pthread_mutex_lock(&ce->lock);
    commit_advance_locked(ce);  /* re-derives majority with the new durable cap;
                                 * broadcasts ce->cond on any advance */
    /* Broadcast even when committed_lsn did not advance: the durable watermark
     * moved, so a waiter's wait_done predicate (which the durable cap feeds via
     * commit_advance_locked) must be re-checked. commit_advance_locked only
     * broadcasts on a committed_lsn change, so cover the no-change case here. */
    pthread_cond_broadcast(&ce->cond);
    pthread_mutex_unlock(&ce->lock);
}

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

    /* Register the durable-advance wake so the flusher can re-trigger commit
     * evaluation when durable_lsn moves (task 2.2). Persist owns the hook slot;
     * we only register a function + ctx, preserving the commit→persist (never
     * the reverse) dependency direction. */
    persist_set_durable_cb(p, commit_on_durable_advance, ce);
    return 0;
}

void commit_engine_fini(commit_engine_t* ce)
{
    if (ce == NULL) {
        return;
    }
    /* Unregister the durable-advance wake BEFORE tearing down the mutex/cond:
     * otherwise the flusher could fire commit_on_durable_advance against a
     * destroyed engine (use-after-free / ASAN). Callers are expected to have
     * stopped the flusher (persist_stop) first, but unhooking is defensive and
     * cheap. After this, persist holds no pointer to ce. */
    if (ce->p != NULL) {
        persist_set_durable_cb(ce->p, NULL, NULL);
    }
    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);
}