/*
 * 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.
 *
 * raft-apply-decoupling task 1.3: commit_submit_async() registers a
 * continuation instead of parking a caller thread, which removes the ONLY
 * mechanism (the caller's own pthread_cond_timedwait deadline) that ever
 * detected a commit timing out. commit_engine_t therefore owns a dedicated
 * timeout-sweep thread (commit_sweep_worker, started in commit_engine_init /
 * stopped+joined in commit_engine_fini) that sleeps on the SAME ce->cond
 * used everywhere else in this file, waking either on a broadcast (new
 * registration, majority advance, membership change, gate change) or at the
 * earliest pending continuation's deadline, whichever comes first.
 *
 * WHY A DEDICATED THREAD, NOT THE SHARED CONTROLLER delaymgr: this task's
 * boundary is commit.c/commit.h only (no cluster.c/ha.c wiring changes), and
 * the shared delaymgr's production polling precision (200ms, cluster.c) is
 * coarse relative to DEFAULT_SYNC_TIMEOUT_MS (500ms) — routing through it
 * would measurably loosen the "identical failure timing" bar (Req 11.1).
 * Reusing ce->lock/ce->cond and the exact wait_deadline() computation
 * commit_submit's blocking path already uses gives the async path the same
 * precision as the blocking path it replaces, with a single thread
 * regardless of how many continuations are pending (unlike the old
 * one-parked-thread-per-waiter model). This thread is private to the commit
 * engine and never touches iomgr or delaymgr, so it does not interact with
 * either thread's non-blocking contract (.kiro/steering/ha-async-io.md).
 *
 * 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 <stdlib.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

/* Forward declaration: raft-apply-decoupling task 1.2. commit_on_durable_advance
 * (below) is defined ahead of the Continuation Resolution section further down
 * this file but must call commit_resolve_pending after releasing ce->lock; see
 * that section for the full definition and contract. */
static void commit_resolve_pending(commit_engine_t* ce);

/* Forward declaration: raft-apply-decoupling task 1.3. commit_engine_init
 * (below) starts the timeout-sweep thread; its entry function is defined in
 * the Timeout Sweep section further down this file (alongside
 * commit_sweep_scan_locked, the locked-scan helper it drives). */
static void* commit_sweep_worker(void* arg);

/* Forward declaration: raft-apply-decoupling task 1.4. commit_engine_init
 * (below) registers this as persist's truncation-wake callback; it is
 * defined in the Continuation Resolution section further down this file
 * (alongside commit_resolve_pending_above, the helper it delegates to). */
static void commit_on_truncate(void* ctx, uint64_t cutoff_lsn);

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

    /* ha-term-safety commit-ceiling-optA: the durable-majority barrier (this
     * function) PLUS the own-term writable gate (commit_set_writable_lsn /
     * wait_done_locked) are authoritative for commit-advance. The raft library's
     * commit_index is NOT folded in as a ceiling here.
     *
     * Why the former min(durable-majority, raft_commit_index) ceiling was
     * removed: under single-writer (only the current primary appends) the
     * durable-majority candidate is ALREADY current-term-safe — every entry it
     * can cover is one this primary appended in its own term, and the own-term
     * NOOP gate (writable_lsn) independently blocks acknowledging a prior-term
     * entry until the current-term NOOP is majority-durable. The library
     * commit_index is computed from match_index[] that the catch-up/streaming
     * path advances asynchronously, so it routinely LAGS the durable majority;
     * folding it into `cap` pinned committed_lsn below the safe durable
     * watermark (e.g. committed stuck at 1 while a 3-node cluster is durable to
     * 5), a redundant, starvation-prone gate. commit_set_raft_commit_index is
     * now inert-advisory (recorded for observability/conformance only). */

    /* Standalone / single-member: self constitutes the majority (Req 6.3) —
     * self still only counts up to what is durable locally, so a single-member
     * commit 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);
    /* INFO (not TRACE): the authoritative primary committed_lsn advance. Fires
     * only on a real advance (monotonic guard above), so it is not noisy, and it
     * is the canonical operator/test signal for the committed watermark — the
     * standby-reported committed_lsn in ha_process_standby_state heartbeats lags
     * and only logs on the one-time converged transition. */
    INFO("commit: committed_lsn %lu -> %lu (n_peers=%d)", cur, cand,
         ce->n_peers);
    pthread_cond_broadcast(&ce->cond);
}

/* Admission cap for concurrent blocking waits, LOCAL (no-pool) fallback:
 * keep COMMIT_WAIT_RESERVED vtasks free of blocking commit waits, floored at 1
 * so one wait is always admittable (Req 7.1, 7.2). Used only when no vtaskmgr
 * pool is wired (commit_set_pool); when a pool is wired the pool owns the
 * single reservation (cap = n_vcpus - VTASK_LIVENESS_RESERVED). */
static int wait_cap(const commit_engine_t* ce)
{
    int cap = ce->n_vcpus - COMMIT_WAIT_RESERVED;

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

/* raft-drive-async task 1.6: acquire a blocking-commit-wait admission slot.
 * When a pool is wired, delegate to the pool's single LivenessAdmission
 * reservation (the pool owns the cap); otherwise apply the local cap. Returns
 * 0 (admitted) or -XCE_ERR_NOT_READY (cap exhausted — fast reject, never
 * blocks). Caller holds ce->lock. inflight_waits is maintained in BOTH modes
 * as the observable in-flight commit-wait count. */
static int wait_admit_locked(commit_engine_t* ce)
{
    if (ce->pool != NULL) {
        int rc = vtaskmgr_blocking_acquire(ce->pool);
        if (rc != 0) {
            return rc; /* -E_XCE_NOT_READY == -XCE_ERR_NOT_READY */
        }
        ce->inflight_waits++;
        return 0;
    }

    /* Local fallback: cap = max(1, n_vcpus - COMMIT_WAIT_RESERVED). */
    if (ce->inflight_waits >= wait_cap(ce)) {
        return -XCE_ERR_NOT_READY;
    }
    ce->inflight_waits++;
    return 0;
}

/* Release a blocking-commit-wait admission slot. Caller holds ce->lock. */
static void wait_release_locked(commit_engine_t* ce)
{
    ce->inflight_waits--;
    if (ce->pool != NULL) {
        vtaskmgr_blocking_release(ce->pool);
    }
}

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

/* raft-apply-decoupling task 1.3: three-way compare of two CLOCK_REALTIME
 * timespecs (the same clock wait_deadline() uses). Used by the timeout-sweep
 * thread to detect an expired deadline and to track the earliest of several
 * pending ones. */
static int ts_cmp(const struct timespec* a, const struct timespec* b)
{
    if (a->tv_sec != b->tv_sec) {
        return (a->tv_sec < b->tv_sec) ? -1 : 1;
    }
    if (a->tv_nsec != b->tv_nsec) {
        return (a->tv_nsec < b->tv_nsec) ? -1 : 1;
    }
    return 0;
}

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

    /* raft-apply-decoupling task 1.2: the durable watermark moving can newly
     * satisfy a registered continuation's predicate exactly like it does a
     * blocked COMMIT_MODE_WAIT caller's -- resolve after ce->lock is released. */
    commit_resolve_pending(ce);
}

/* ============================================================
 * 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->pool           = NULL; /* local cap until commit_set_pool wires the pool */
    ce->inflight_waits = 0;
    ce->writable_lsn   = 0; /* no own-term gate until set by promotion */
    ce->raft_commit_index = 0; /* advisory record only; does not gate commit (optA) */
    memset(ce->scratch, 0, sizeof(ce->scratch));

    /* raft-apply-decoupling task 1.1: continuation registry, sized to
     * n_vcpus (see the field comment in commit.h for the sizing invariant:
     * wait_cap(ce) <= n_vcpus always, so n_vcpus slots suffice regardless
     * of the blocking-WAIT/async-registration mix sharing inflight_waits).
     * calloc zero-fills every slot's lsn to 0, the free-slot sentinel. */
    ce->continuations = (commit_continuation_t*)calloc(
        (size_t)n_vcpus, sizeof(commit_continuation_t));
    if (ce->continuations == NULL) {
        pthread_cond_destroy(&ce->cond);
        pthread_mutex_destroy(&ce->lock);
        return -ENOMEM;
    }
    ce->n_continuations = n_vcpus;

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

    /* raft-apply-decoupling task 1.4: register the truncation wake alongside
     * the durable-advance wake -- same dependency direction (persist stores
     * an opaque callback; commit.c is never a dependency of persist.c). Fired
     * by persist_truncate_wal_above after it actually removes entries above
     * a new tail, so any continuation registered above that tail (which can
     * never commit -- its entry no longer exists) resolves FAILED promptly. */
    persist_set_truncate_cb(p, commit_on_truncate, ce);

    /* raft-apply-decoupling task 1.3: start the dedicated timeout-sweep
     * thread (see file header for why this is a private thread rather than
     * the shared controller delaymgr). A creation failure is treated as a
     * hard init failure -- like the calloc failure above -- because a
     * commit engine that cannot detect async timeouts cannot honor its
     * documented contract (Req 9.4: every registered continuation must
     * eventually resolve). */
    ce->sweep_stop   = 0;
    ce->sweep_thread = 0; /* explicit sentinel: pthread_create leaves this
                           * unspecified on failure (POSIX), and ce may be
                           * caller-reused memory from a prior lifecycle. */
    {
        int prc = pthread_create(&ce->sweep_thread, NULL, commit_sweep_worker, ce);
        if (prc != 0) {
            persist_set_durable_cb(p, NULL, NULL);
            persist_set_truncate_cb(p, NULL, NULL);
            free(ce->continuations);
            ce->continuations   = NULL;
            ce->n_continuations = 0;
            pthread_cond_destroy(&ce->cond);
            pthread_mutex_destroy(&ce->lock);
            return -prc;
        }
    }
    return 0;
}

/* raft-drive-async task 1.6: fold commit-wait admission onto the vtaskmgr
 * pool's single LivenessAdmission reservation. Set-once at wiring time; safe
 * to set NULL to restore the local fallback. The pool must outlive the engine
 * (controller lifecycle guarantees this — vtaskmgr_fini after commit_engine_fini). */
void commit_set_pool(commit_engine_t* ce, vtaskmgr_t* pool)
{
    if (ce == NULL) {
        return;
    }
    pthread_mutex_lock(&ce->lock);
    ce->pool = pool;
    pthread_mutex_unlock(&ce->lock);
}

void commit_engine_fini(commit_engine_t* ce)
{
    if (ce == NULL) {
        return;
    }
    /* raft-apply-decoupling task 1.3: stop + join the timeout-sweep thread
     * BEFORE tearing down lock/cond/continuations -- it reads all three.
     * Signal under the lock and broadcast to wake it from either
     * pthread_cond_wait (no pending continuations) or pthread_cond_timedwait
     * (a pending deadline); the join guarantees it has fully exited before
     * anything it touches is freed or destroyed. */
    if (ce->sweep_thread != 0) {
        pthread_mutex_lock(&ce->lock);
        ce->sweep_stop = 1;
        pthread_cond_broadcast(&ce->cond);
        pthread_mutex_unlock(&ce->lock);
        pthread_join(ce->sweep_thread, NULL);
        ce->sweep_thread = 0;
    }
    /* 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);
        persist_set_truncate_cb(ce->p, NULL, NULL);
    }
    free(ce->continuations);
    ce->continuations   = NULL;
    ce->n_continuations = 0;
    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             wrc, arc;

    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. When a vtaskmgr pool is wired (raft-drive-async task 1.6)
     * the cap is the pool's single LivenessAdmission reservation; otherwise
     * the local COMMIT_WAIT_RESERVED cap. */
    arc = wait_admit_locked(ce);
    if (arc != 0) {
        pthread_mutex_unlock(&ce->lock);
        WARN("commit: WAIT rejected, admission cap exhausted "
             "(pool=%s n_vcpus=%d lsn=%lu)",
             ce->pool ? "pool" : "local", ce->n_vcpus, lsn);
        return arc;
    }
#ifdef XCE_TESTING
    assert(ce->inflight_waits > 0);
#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;
        }
    }

    wait_release_locked(ce);
#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);

    /* raft-apply-decoupling task 1.2: the n_peers<=1 self-quorum branch above
     * can advance committed_lsn via this same lock section -- resolve any
     * newly-satisfied registered continuation after ce->lock is released. */
    commit_resolve_pending(ce);

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

/* Caller holds ce->lock. Returns the index of a free (lsn == 0) slot, or -1
 * if none is free. n_continuations == n_vcpus >= wait_cap(ce) >=
 * inflight_waits at all times (the same admission cap governs both this
 * registry and COMMIT_MODE_WAIT's blocking parks, sharing one budget), so an
 * admitted registration always finds a free slot -- see the commit.h field
 * comment for the sizing invariant. */
static int continuation_find_free_slot_locked(const commit_engine_t* ce)
{
    for (int i = 0; i < ce->n_continuations; i++) {
        if (ce->continuations[i].lsn == 0) {
            return i;
        }
    }
    return -1;
}

/* raft-apply-decoupling task 1.1: event-driven registration, the
 * non-blocking sibling of commit_submit(..., COMMIT_MODE_WAIT, ...).
 * Registers `fn`/`ctx` against `lsn` and returns immediately -- it never
 * evaluates or waits on the commit predicate itself. Resolution (invoking
 * `fn` on majority-ack / timeout / demotion / truncation) is
 * commit_resolve's job, wired onto commit_advance_locked and the existing
 * failure paths by a follow-up task; until then a registered slot is
 * inert. */
int commit_submit_async(commit_engine_t* ce, uint64_t lsn,
                        commit_continuation_fn fn, void* ctx)
{
    int slot, arc;

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

    pthread_mutex_lock(&ce->lock);

    /* Same admission-cap discipline as COMMIT_MODE_WAIT (Req 9.1, 9.2): a
     * registered continuation occupies the identical bounded slot a
     * blocking wait would have parked in, so the two calling conventions
     * share one budget while both exist side by side. */
    arc = wait_admit_locked(ce);
    if (arc != 0) {
        pthread_mutex_unlock(&ce->lock);
        WARN("commit: async registration rejected, admission cap exhausted "
             "(pool=%s n_vcpus=%d lsn=%lu)",
             ce->pool ? "pool" : "local", ce->n_vcpus, lsn);
        return arc;
    }

    slot = continuation_find_free_slot_locked(ce);
    if (slot < 0) {
        /* Unreachable given the sizing invariant above; fail loud rather
         * than silently drop the registration if it is ever violated. */
        wait_release_locked(ce);
        pthread_mutex_unlock(&ce->lock);
        WARN("commit: async registration rejected, no free continuation "
             "slot (lsn=%lu, n_continuations=%d) -- sizing invariant "
             "violated", lsn, ce->n_continuations);
        return -XCE_ERR_NOT_READY;
    }

    ce->continuations[slot].lsn = lsn;
    ce->continuations[slot].fn  = fn;
    ce->continuations[slot].ctx = ctx;
    /* raft-apply-decoupling task 1.3: arm this registration's timeout with
     * the SAME deadline computation and SAME default constant
     * commit_submit's blocking WAIT path uses, so an async registration
     * observes identical failure timing to today's blocking path (Req
     * 11.1). The dedicated timeout-sweep thread (commit_sweep_worker)
     * resolves this continuation FAILED/-ETIMEDOUT if it is still pending
     * once this deadline passes. */
    wait_deadline(&ce->continuations[slot].deadline, DEFAULT_SYNC_TIMEOUT_MS);

    /* Wake the sweep thread: a freshly armed deadline may be sooner than
     * whatever it is currently sleeping toward. */
    pthread_cond_broadcast(&ce->cond);
    pthread_mutex_unlock(&ce->lock);
    return 0;
}

/* ============================================================
 * Continuation resolution (raft-apply-decoupling task 1.2)
 * ============================================================ */

/* Heap payload for one dispatched continuation invocation. Owned by the
 * enqueued vtask's vtask_data_t.u.generic; vtask_fini/run_task only free the
 * vtask_t and its vtask_data_t wrapper, never this payload, so the dispatch
 * entry function below frees it after invoking fn. */
typedef struct {
    commit_continuation_fn fn;
    void*                  ctx;
    uint64_t               lsn;
    commit_outcome_t       outcome;
    int                    rc;
} commit_resolve_dispatch_t;

/* vtask entry (run on a pool worker, never the calling thread): invoke the
 * continuation and free its dispatch payload. */
static int commit_resolve_dispatch_entry(vtask_t* task)
{
    commit_resolve_dispatch_t* d =
        (commit_resolve_dispatch_t*)task->data->u.generic;

    d->fn(d->ctx, d->lsn, d->outcome, d->rc);
    free(d);
    return 0;
}

/* Build + enqueue a generic vtask that invokes fn(ctx, lsn, outcome, rc) off
 * the calling thread. Caller must NOT hold ce->lock.
 *
 * Class VTASK_LIVE_BLOCK, not VTASK_REQ_BLOCK or VTASK_LIVENESS:
 *   - VTASK_REQ_BLOCK is admission-gated (vtask_enq_classified can fast-reject
 *     with -E_XCE_NOT_READY at the pool's blocking cap) -- unacceptable here:
 *     Req 9.4 requires the continuation be resolved exactly once, never left
 *     unresolved, and a rejected enqueue would silently drop it.
 *   - VTASK_LIVENESS is never admission-gated either, but its reservation
 *     (vtaskmgr.h) is sized and reserved for the narrow WAL-ack/heartbeat/
 *     vote receipt+send set -- not general apply/notify continuations.
 *   - VTASK_LIVE_BLOCK is never admission-gated (same delivery guarantee as
 *     LIVENESS) and is documented as "bounded mandatory park ... runs in the
 *     reserved headroom" -- the correct fit for mandatory, must-not-reject
 *     work that is not itself the narrow liveness set.
 *
 * On allocation/enqueue failure (OOM, or no vtaskmgr pool wired -- true only
 * of commit-only unit fixtures; production always wires the pool before any
 * commit activity, cluster_init_phase2) this fails loud and invokes the
 * continuation inline as a last resort: "controllers never self-terminate" --
 * a continuation must never be silently dropped, so a rare fallback inline
 * call is preferred over an unresolved continuation. */
static void dispatch_continuation(commit_engine_t* ce, commit_continuation_fn fn,
                                  void* ctx, uint64_t lsn,
                                  commit_outcome_t outcome, int rc)
{
    vtask_data_t*              data;
    vtask_t*                   task;
    commit_resolve_dispatch_t* d;

    if (fn == NULL) {
        return;
    }

    data = calloc(1, sizeof(*data));
    task = vtask_alloc();
    d    = calloc(1, sizeof(*d));

    if (ce->pool == NULL || data == NULL || task == NULL || d == NULL) {
        WARN("commit: continuation dispatch unavailable (pool=%p data=%p "
             "task=%p d=%p) for lsn=%lu outcome=%d rc=%d -- invoking inline "
             "(stay alive, never leave a continuation unresolved)",
             (void*)ce->pool, (void*)data, (void*)task, (void*)d, lsn,
             (int)outcome, rc);
        free(data);
        free(task);
        free(d);
        fn(ctx, lsn, outcome, rc);
        return;
    }

    d->fn      = fn;
    d->ctx     = ctx;
    d->lsn     = lsn;
    d->outcome = outcome;
    d->rc      = rc;

    data->vtype     = VTASK_GENERIC;
    data->u.generic = d;
    vtask_init(task, (vtask_func_t)commit_resolve_dispatch_entry, data);

    /* VTASK_LIVE_BLOCK is never admission-gated -- see the function comment
     * above; this cannot fast-reject the way VTASK_REQ_BLOCK could. */
    if (vtask_enq_classified(ce->pool, task, VTASK_LIVE_BLOCK) < 0) {
        WARN("commit: continuation vtask enqueue failed for lsn=%lu -- "
             "invoking inline (stay alive)", lsn);
        free(data);
        free(task);
        free(d);
        fn(ctx, lsn, outcome, rc);
    }
}

/* Resolve lsn's registered continuation (if any) exactly once: find + clear
 * its slot under ce->lock (idempotent -- a cleared slot's lsn is 0 and never
 * matches again), release its admission hold (same shared budget as
 * COMMIT_MODE_WAIT), then dispatch off-thread AFTER ce->lock is released
 * (lock-ordering rule, design §Commit Engine Resolve). Caller must NOT hold
 * ce->lock. */
void commit_resolve(commit_engine_t* ce, uint64_t lsn, commit_outcome_t outcome,
                    int rc)
{
    commit_continuation_fn fn  = NULL;
    void*                   ctx = NULL;

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

    pthread_mutex_lock(&ce->lock);
    for (int i = 0; i < ce->n_continuations; i++) {
        if (ce->continuations[i].lsn == lsn) {
            fn  = ce->continuations[i].fn;
            ctx = ce->continuations[i].ctx;
            ce->continuations[i].lsn = 0; /* free the slot -- idempotency */
            ce->continuations[i].fn  = NULL;
            ce->continuations[i].ctx = NULL;
            wait_release_locked(ce);
            break;
        }
    }
    pthread_mutex_unlock(&ce->lock);

    if (fn != NULL) {
        dispatch_continuation(ce, fn, ctx, lsn, outcome, rc);
    }
}

/* Scan the continuation registry for entries whose WAIT-equivalent completion
 * predicate (wait_done_locked -- own entry committed AND the own-term gate,
 * when set, reached) is now satisfied, and resolve each with COMMITTED, rc=0
 * -- the async-registration mirror of what a woken COMMIT_MODE_WAIT caller's
 * own re-check observes. Caller must NOT hold ce->lock: this takes the lock
 * itself for a bounded snapshot scan, then resolves each candidate from an
 * unlocked context via the self-locking commit_resolve() (satisfying the
 * dispatch-after-unlock ordering rule). Safe under races: a snapshotted lsn
 * already resolved by a concurrent caller by the time commit_resolve() runs
 * is simply a no-op (idempotent), so no continuation is ever double-resolved. */
static void commit_resolve_pending(commit_engine_t* ce)
{
    int n = 0;

    if (ce == NULL) {
        return;
    }

    uint64_t ready[ce->n_continuations]; /* VLA: bounded by n_vcpus -- see the
                                          * commit.h field-comment sizing
                                          * invariant */

    pthread_mutex_lock(&ce->lock);
    for (int i = 0; i < ce->n_continuations; i++) {
        uint64_t lsn = ce->continuations[i].lsn;

        if (lsn != 0 && wait_done_locked(ce, lsn)) {
            ready[n++] = lsn;
        }
    }
    pthread_mutex_unlock(&ce->lock);

    for (int i = 0; i < n; i++) {
        commit_resolve(ce, ready[i], COMMIT_OUTCOME_COMMITTED, 0);
    }

    /* raft-apply-decoupling task 6.1: this function is the single choke
     * point every one of the six committed_lsn-advancing paths already
     * funnels through (durable-advance, self-quorum commit_submit, ack-
     * driven advance, the writable-lsn gate, the raft-library commit-index
     * feed via commit_set_raft_commit_index, and membership change) --
     * kicking the ordered apply pump here, rather than only from ha.c's
     * on_commit_advanced callback as design.md's wiring text names, is a
     * deliberate deviation: on_commit_advanced is fed by exactly ONE of
     * these six paths, so wiring the pump there alone would silently starve
     * it for any cluster (e.g. single-node/self-quorum) that never
     * exercises that specific raft-library callback. Called unconditionally
     * -- persist_apply_pump_kick is itself a documented no-op whenever
     * ce->p->cluster is NULL, so this has zero effect on any commit-engine
     * unit fixture that never wires a cluster_t (the overwhelming majority
     * of this file's own test suite). */
    persist_apply_pump_kick(ce->p, ce->pool);
}

/* raft-apply-decoupling task 1.4: the FAILED-outcome sibling of
 * commit_resolve_pending -- instead of scanning for majority-satisfied
 * entries, resolves every registered continuation whose lsn > lsn_floor as
 * FAILED with rc. See commit.h for the full contract and the two intended
 * callers (log-tail truncation, demotion). Caller must NOT hold ce->lock. */
void commit_resolve_pending_above(commit_engine_t* ce, uint64_t lsn_floor,
                                  int rc)
{
    int n = 0;

    if (ce == NULL) {
        return;
    }

    uint64_t failed[ce->n_continuations]; /* VLA: bounded by n_vcpus -- see
                                           * the commit.h field-comment
                                           * sizing invariant */

    pthread_mutex_lock(&ce->lock);
    for (int i = 0; i < ce->n_continuations; i++) {
        uint64_t lsn = ce->continuations[i].lsn;

        if (lsn != 0 && lsn > lsn_floor) {
            failed[n++] = lsn;
        }
    }
    pthread_mutex_unlock(&ce->lock);

    for (int i = 0; i < n; i++) {
        commit_resolve(ce, failed[i], COMMIT_OUTCOME_FAILED, rc);
    }
}

/* Truncation wake (raft-apply-decoupling task 1.4): registered with persist
 * via persist_set_truncate_cb (commit_engine_init) and invoked by
 * persist_truncate_wal_above AFTER it actually removes entries above the new
 * tail (cutoff_lsn) and AFTER it releases p->lock -- persist only fires this
 * when a truncation genuinely happened (current_lsn strictly decreased); a
 * no-op truncate never calls this. Every continuation registered above the
 * new tail can never commit (its entry no longer exists in the log) --
 * resolve it FAILED now rather than leaving it for the timeout-sweep thread
 * to eventually time out. -ENOENT mirrors the truncation-outcome vocabulary
 * already exercised in test_commit_resolve.c (task 1.2).
 *
 * LOCK ORDERING: mirrors commit_on_durable_advance -- runs with NO persist
 * lock held, takes only ce->lock (via commit_resolve_pending_above /
 * commit_resolve). Same cycle-free p->lock -> ce->lock edge; the commit side
 * never takes p->lock while holding ce->lock. ctx is the commit_engine_t
 * registered in commit_engine_init. */
static void commit_on_truncate(void* ctx, uint64_t cutoff_lsn)
{
    commit_engine_t* ce = ctx;

    if (ce == NULL || ce->p == NULL) {
        return;
    }
    commit_resolve_pending_above(ce, cutoff_lsn, -ENOENT);
}

/* ============================================================
 * Timeout sweep (raft-apply-decoupling task 1.3)
 * ============================================================ */

/* One bounded, read-only classification pass over the continuation
 * registry: for every still-registered lsn, majority-satisfied ALWAYS wins
 * over an expired deadline when both are true in this same lock hold --
 * mirroring commit_submit's own last-look-before-timeout tie-break (a
 * majority ack landing exactly at the deadline is never spuriously reported
 * as a timeout). Populates committed_ready[] and failed_ready[] (via nc and
 * nf) and reports the earliest deadline among whatever remains pending via
 * next_deadline and have_next. Caller MUST hold ce->lock and does its own
 * unlocking -- this function never blocks and never dispatches, so it is
 * safe to call as the last thing done before committing to a
 * pthread_cond_wait/pthread_cond_timedwait in the SAME lock hold (see
 * commit_sweep_worker). */
static void commit_sweep_scan_locked(commit_engine_t* ce,
                                     uint64_t* committed_ready, int* nc,
                                     uint64_t* failed_ready, int* nf,
                                     struct timespec* next_deadline,
                                     int* have_next)
{
    struct timespec now;

    clock_gettime(CLOCK_REALTIME, &now); /* same clock as wait_deadline() */
    *nc        = 0;
    *nf        = 0;
    *have_next = 0;

    for (int i = 0; i < ce->n_continuations; i++) {
        uint64_t lsn = ce->continuations[i].lsn;

        if (lsn == 0) {
            continue;
        }
        if (wait_done_locked(ce, lsn)) {
            committed_ready[(*nc)++] = lsn;
        } else if (ts_cmp(&now, &ce->continuations[i].deadline) >= 0) {
            failed_ready[(*nf)++] = lsn;
        } else if (!*have_next ||
                  ts_cmp(&ce->continuations[i].deadline, next_deadline) < 0) {
            *next_deadline = ce->continuations[i].deadline;
            *have_next     = 1;
        }
    }
}

/* Dedicated timeout-sweep thread, private to this commit_engine_t (see the
 * file header for why this is not routed through the shared controller
 * delaymgr). Sleeps on ce->cond to the earliest pending continuation
 * deadline, or indefinitely when none are pending; a broadcast (new
 * registration, majority advance, membership change, gate change, or
 * commit_engine_fini's shutdown signal) wakes it early to re-evaluate.
 * Exits once ce->sweep_stop is observed under ce->lock.
 *
 * LOST-WAKEUP FIX: commit_sweep_scan_locked() only classifies -- it never
 * dispatches -- so every dispatch (commit_resolve(), which must run
 * UNLOCKED per the lock-ordering rule) happens between an unlock and the
 * next lock in this loop. A commit_submit_async() registration (or any
 * other broadcaster; all of them broadcast while HOLDING ce->lock) landing
 * during that unlocked window would find no thread parked in cond_wait to
 * receive it -- POSIX condvars do not latch a signal sent to no waiter.
 * The fix follows the standard "recheck the wait condition in the SAME
 * lock hold that leads directly into the wait call" idiom: whenever this
 * loop has something to dispatch, it unlocks, dispatches, RE-LOCKS, and
 * loops back to scan again -- it never commits to a wait based on a
 * have_next/next_deadline computed before an unlocked dispatch step. Only
 * a scan that found NOTHING to dispatch (so no unlock is needed before the
 * wait) is trusted to size the wait. */
static void* commit_sweep_worker(void* arg)
{
    commit_engine_t* ce = (commit_engine_t*)arg;

    pthread_mutex_lock(&ce->lock);
    for (;;) {
        uint64_t        committed_ready[ce->n_continuations]; /* VLA: bounded
                                                                * by n_vcpus,
                                                                * see the
                                                                * commit.h
                                                                * sizing
                                                                * invariant */
        uint64_t        failed_ready[ce->n_continuations];
        int             nc, nf, have_next;
        struct timespec next_deadline;

        if (ce->sweep_stop) {
            break;
        }

        commit_sweep_scan_locked(ce, committed_ready, &nc, failed_ready, &nf,
                                 &next_deadline, &have_next);

        if (nc > 0 || nf > 0) {
            pthread_mutex_unlock(&ce->lock);
            for (int i = 0; i < nc; i++) {
                commit_resolve(ce, committed_ready[i], COMMIT_OUTCOME_COMMITTED, 0);
            }
            for (int i = 0; i < nf; i++) {
                WARN("commit: async continuation deadline expired without "
                     "commit (lsn=%lu committed_lsn=%lu writable_lsn=%lu "
                     "timeout_ms=%u)",
                     failed_ready[i], atomic_load(&ce->p->committed_lsn),
                     ce->writable_lsn, (unsigned)DEFAULT_SYNC_TIMEOUT_MS);
                commit_resolve(ce, failed_ready[i], COMMIT_OUTCOME_FAILED,
                               -ETIMEDOUT);
            }
            pthread_mutex_lock(&ce->lock);
            continue; /* re-scan under the lock before ever waiting */
        }

        /* Nothing to dispatch: this scan just ran in the exact lock hold we
         * are about to wait under, so no broadcast between "now" and the
         * wait call below can be missed. */
        if (have_next) {
            pthread_cond_timedwait(&ce->cond, &ce->lock, &next_deadline);
        } else {
            pthread_cond_wait(&ce->cond, &ce->lock);
        }
    }
    pthread_mutex_unlock(&ce->lock);

    return NULL;
}

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

    /* raft-apply-decoupling task 1.2: an ACK-driven advance can newly satisfy
     * a registered continuation -- resolve after ce->lock is released. */
    commit_resolve_pending(ce);
}

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

    /* raft-apply-decoupling task 1.2: the gate opening can newly satisfy a
     * registered continuation whose own entry already committed. */
    commit_resolve_pending(ce);
}

/* ha-term-safety commit-ceiling-optA: record the raft library's commit_index
 * for OBSERVABILITY / TRACE / conformance-oracle purposes ONLY. This value does
 * NOT gate commit-advance — majority_candidate_locked no longer folds it into
 * the commit candidate (Option A: the durable-majority barrier + own-term
 * writable gate are authoritative). It is still maintained (and still fed by
 * on_commit_advanced) so the library half keeps a sink and tooling can compare
 * the library's current-term commit point against the barrier's committed_lsn.
 * Monotonic max() — never regresses.
 *
 * The barrier is still re-evaluated + waiters woken here: the on_commit_advanced
 * callback fires on the ACK path, and a fresh ACK may have just raised the
 * durable majority, so this is a convenient wake point (a no-op if nothing
 * changed). commit_advance_locked re-derives the durable majority WITHOUT the
 * recorded commit_index.
 *
 * LOCK ORDERING: takes only ce->lock (commit_advance_locked re-reads the durable
 * watermark + registry watermarks via the existing path). The on_commit_advanced
 * callback runs on the iomgr ACK path that drove raft_advance_commit; that path
 * does not hold p->lock here, preserving the p->lock → ce->lock direction. */
void commit_set_raft_commit_index(commit_engine_t* ce, uint64_t commit_index)
{
    if (ce == NULL) {
        return;
    }
    pthread_mutex_lock(&ce->lock);
    if (commit_index > ce->raft_commit_index) {
        ce->raft_commit_index = commit_index; /* monotonic advisory record */
    }
    /* Advisory record updated. Re-derive the durable majority (which does NOT
     * consult raft_commit_index) and wake waiters in case a concurrent ACK on
     * this path raised the durable watermark. */
    commit_advance_locked(ce);
    pthread_cond_broadcast(&ce->cond);
    pthread_mutex_unlock(&ce->lock);

    /* raft-apply-decoupling task 1.2: resolve after ce->lock is released. */
    commit_resolve_pending(ce);
}

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

    /* raft-apply-decoupling task 1.2: resolve after ce->lock is released. */
    commit_resolve_pending(ce);
}