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

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

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

    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;

    pthread_mutex_unlock(&ce->lock);
    return 0;
}

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

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

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