myxce/src/controller/repl.c

/*
 * repl.c — P2-1: Replicator lifecycle, slot management, and WAL trim guard.
 *
 * mylock_t is a plain pthread_mutex_t alias (see mylock.h).
 * API: mylock_init / mylock_fini / mylock_lock / mylock_unlock
 */
#include <assert.h>
#include <errno.h>
#include <string.h>

#include "controller.capnp.h"  /* OPCode_walAck, OPCode_upgradeConfigAck */
#include "repl.h"
#include "commit.h"    /* commit engine feed + adapter (ha-commit-engine 5.2) */
#include "mylock.h"
#include "utils.h"
#include "request.h"   /* xce_wal_batch_req_raw */
#include "typemap.h"   /* wal_catchup_entry_t */

#ifdef XCE_TESTING
/* Test-injectable send hook: replaces xce_wal_batch_req_raw in unit tests. */
int (*repl_broadcast_send_hook)(void *ep,
                                wal_catchup_entry_t **entries,
                                int n_entries) = NULL;
#endif

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

int repl_init(replicator_t *r)
{
    if (r == NULL) {
        return -EINVAL;
    }

    memset(r, 0, sizeof(*r));

    int rc = mylock_init(&r->lock);
    if (rc != 0) {
        return rc;
    }

    rc = pthread_mutex_init(&r->ack_lock, NULL);
    if (rc != 0) {
        mylock_fini(&r->lock);
        return -rc;
    }

    rc = pthread_cond_init(&r->ack_cond, NULL);
    if (rc != 0) {
        pthread_mutex_destroy(&r->ack_lock);
        mylock_fini(&r->lock);
        return -rc;
    }

    return 0;
}

void repl_fini(replicator_t *r)
{
    if (r == NULL) {
        return;
    }

    pthread_cond_destroy(&r->ack_cond);
    pthread_mutex_destroy(&r->ack_lock);
    mylock_fini(&r->lock);
}

/* ============================================================
 * Standby slot management
 * ============================================================ */

int repl_add_standby(replicator_t *r, repl_ep_t ep, const char *host, uint32_t port)
{
    if (r == NULL || ep == NULL || host == NULL) {
        return -EINVAL;
    }

    mylock_lock(&r->lock);

    /* Idempotency guard: if ep is already registered, return its slot index.
     * Req 1.3: duplicate repl_add_standby calls are treated as no-ops. */
    for (int i = 0; i < MAX_STANDBYS; i++) {
        if (r->standbys[i].ep == ep) {
            mylock_unlock(&r->lock);
            return i;
        }
    }

    if (r->nstandbys >= MAX_STANDBYS) {
        mylock_unlock(&r->lock);
        return -ENOSPC;
    }

    /* Slot allocation (bug-1474 hardening): prefer a genuinely virgin slot
     * over one still holding an unexpired bug-1147 disconnect guard
     * (last_disconnected_lsn != 0) for a DIFFERENT peer. repl_add_standby
     * cannot know whether this call is a legitimate new connection or a
     * duplicate/late-joiner reconnect for an already-connected peer (that
     * classification defect lives in ha.c, out of scope here) — but either
     * way, clobbering an unexpired guard is never correct: it silently
     * defeats repl_trim_lsn's WAL retention floor for whichever standby
     * the guard was protecting (see .wolf/buglog.json bug-1474).
     *
     * Pass 1: any slot with ep==NULL && last_disconnected_lsn==0 (virgin —
     * never used, or its guard already expired via
     * repl_expire_disconnect_slots, which fully zeroes the slot). Always
     * safe to take; first such slot by index.
     *
     * Pass 2 (fallback, only when Pass 1 finds nothing): among slots with
     * ep==NULL && last_disconnected_lsn!=0 (guard-holding), reuse the one
     * with the OLDEST last_disconnect_time. Deliberate policy: that guard
     * is closest to its own natural expiry (repl_expire_disconnect_slots'
     * snap_interval_s window), so reusing it discards the LEAST remaining
     * WAL-retention protection of any guard-holding slot. With
     * MAX_STANDBYS==3 it is possible — under a chaotic multi-disconnect
     * scenario with no virgin slot free — for every non-primary slot to be
     * guard-holding simultaneously; refusing this connect (-ENOSPC) would
     * trade a bounded, already-degraded WAL-retention risk for an outright
     * connection failure, which is worse for cluster availability. This
     * fallback does NOT touch the -ENOSPC path above, which remains
     * reserved for the genuinely-impossible case of more distinct live
     * peers than slots exist. */
    int idx = -1;
    for (int i = 0; i < MAX_STANDBYS; i++) {
        if (r->standbys[i].ep == NULL && r->standbys[i].last_disconnected_lsn == 0) {
            idx = i;
            break;
        }
    }

    /* Pass 1 found nothing -- about to fall back to Pass 2 (guard-holding
     * slot reuse) below. Tracked so the WARN below fires only for the
     * genuine fallback case, not the common virgin-slot path. */
    int used_fallback_slot = (idx < 0);

    if (idx < 0) {
        for (int i = 0; i < MAX_STANDBYS; i++) {
            if (r->standbys[i].ep != NULL) {
                continue;
            }
            if (idx < 0 ||
                r->standbys[i].last_disconnect_time < r->standbys[idx].last_disconnect_time) {
                idx = i;
            }
        }
    }

    if (idx < 0) {
        /* Should not happen given nstandbys guard above, but be safe */
        mylock_unlock(&r->lock);
        return -ENOSPC;
    }

    standby_conn_t *s = &r->standbys[idx];
    if (used_fallback_slot) {
        /* bug-1474 review finding: this silently destroys the slot's
         * WAL-retention guard below (via memset) -- log it loudly, mirroring
         * repl_broadcast's existing "degraded but continuing" WARN
         * convention (e.g. "standby %s:%u evicted", line ~314 above), so the
         * discarded protection is operator-visible instead of silent. */
        WARN("repl_add_standby: reusing guard-holding slot %d (host=%s:%u,"
             " last_disconnected_lsn=%lu) -- no virgin slot available,"
             " WAL-retention protection discarded",
             idx, s->host, s->port, (unsigned long)s->last_disconnected_lsn);
    }
    memset(s, 0, sizeof(*s));
    s->ep   = ep;
    s->port = port;
    strncpy(s->host, host, XCE_MAXNAME_LEN - 1);
    s->host[XCE_MAXNAME_LEN - 1] = '\0';
    atomic_store(&s->acked_lsn, 0);
    atomic_store(&s->acked_config_lsn, 0);
    atomic_store(&s->in_catch_up, 0);

    r->nstandbys++;

    mylock_unlock(&r->lock);

    /* Quorum membership changed: re-evaluate pending commits. Called AFTER
     * releasing r->lock — engine lock order is ce->lock then r->lock
     * (ha-commit-engine task 5.2, Req 4.2). The idempotent early-return
     * above mutates nothing, so it deliberately skips this hook. */
    if (r->ce != NULL) {
        commit_on_membership_change(r->ce);
    }
    return idx;
}

void repl_remove_standby(replicator_t *r, repl_ep_t ep)
{
    int removed = 0;

    if (r == NULL || ep == NULL) {
        return;
    }

    mylock_lock(&r->lock);

    for (int i = 0; i < MAX_STANDBYS; i++) {
        standby_conn_t *s = &r->standbys[i];
        if (s->ep == ep) {
            s->last_disconnected_lsn = atomic_load(&s->acked_lsn);
            s->last_disconnect_time  = time(NULL);
            s->ep = NULL;
            r->nstandbys--;
            removed = 1;
            break;
        }
    }

    mylock_unlock(&r->lock);

    /* Membership shrink: wake pending waiters to re-check their predicates
     * (after r->lock release — engine lock order; ha-commit-engine Req 4.2). */
    if (removed && r->ce != NULL) {
        commit_on_membership_change(r->ce);
    }
}

void repl_expire_disconnect_slots(replicator_t *r, time_t now_s, uint32_t snap_interval_s)
{
    if (r == NULL) {
        return;
    }

    mylock_lock(&r->lock);

    for (int i = 0; i < MAX_STANDBYS; i++) {
        standby_conn_t *s = &r->standbys[i];
        /* Only process disconnected (ep==NULL) slots that still hold a guard LSN */
        if (s->ep == NULL && s->last_disconnected_lsn != 0) {
            /* snap_interval_s==0 means expire immediately regardless of wall clock */
            int expired = (snap_interval_s == 0) ||
                          (now_s >= s->last_disconnect_time + (time_t)snap_interval_s);
            if (expired) {
                memset(s, 0, sizeof(*s));
            }
        }
    }

    mylock_unlock(&r->lock);
}

/* ============================================================
 * ACK handling — call from io_worker read callback ONLY
 * ============================================================ */

void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn)
{
    int slot_idx = -1;

    if (r == NULL || ep == NULL) {
        return;
    }

    mylock_lock(&r->lock);

    for (int i = 0; i < MAX_STANDBYS; i++) {
        standby_conn_t *s = &r->standbys[i];
        if (s->ep == ep) {
            atomic_store(&s->acked_lsn, acked_lsn);
            if (s->pending_ack_count > 0)
                s->pending_ack_count--;
            slot_idx = i;
            break;
        }
    }

    mylock_unlock(&r->lock);

    /* Feed the commit engine AFTER the acked_lsn store and AFTER releasing
     * r->lock — commit_on_ack takes ce->lock then r->lock (majority scan),
     * so calling it under r->lock would invert the engine's lock order.
     * Never blocks (iomgr-safe; ha-commit-engine task 5.2, design §ACK Feed). */
    if (r->ce != NULL && slot_idx >= 0) {
        commit_on_ack(r->ce, slot_idx, acked_lsn);
    }

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

/* ============================================================
 * Replication send (task 7.2)
 *
 * Iterates active standbys; skips empty slots and standbys in catch-up.
 * Evicts standbys whose pending_ack_count exceeds MAX_PENDING_ACKS by marking
 * them in_catch_up=1 (inlined to avoid re-entrant lock acquisition).
 * ============================================================ */

int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
                   const void *payload, uint32_t payload_len,
                   uint64_t prev_log_term)
{
    /* WAL_CHECKPOINT entries must never be replicated */
    assert(hdr != NULL);
    assert(hdr->type != WAL_CHECKPOINT);

    if (r == NULL)
        return -EINVAL;

    mylock_lock(&r->lock);

    if (r->nstandbys == 0) {
        WARN("repl_broadcast: no standbys registered — WAL entry lsn=%lu type=%u"
             " not replicated",
             (unsigned long)hdr->lsn, hdr->type);
        mylock_unlock(&r->lock);
        return 0;
    }

    for (int i = 0; i < MAX_STANDBYS; i++) {
        standby_conn_t *s = &r->standbys[i];

        if (s->ep == NULL)
            continue;

        if (atomic_load_explicit(&s->in_catch_up, memory_order_acquire))
            continue;

        if (s->pending_ack_count > MAX_PENDING_ACKS) {
            /* Inline repl_start_catchup to avoid re-entrant lock acquisition */
            s->join_applied_lsn = atomic_load_explicit(&s->acked_lsn,
                                                        memory_order_acquire);
            atomic_store_explicit(&s->in_catch_up, 1, memory_order_release);
            WARN("repl_broadcast: standby %s:%u evicted (pending_ack_count=%u > %u)"
                 " — triggering catch-up reconnect",
                 s->host, s->port, s->pending_ack_count, MAX_PENDING_ACKS);
            continue;
        }

        /* Build a single-entry AppendEntries and enqueue it for this standby.
         *
         * ha-term-safety task 3.4 (Req 1.3/1.4): carry the leader's per-entry
         * term + the AppendEntries prev tuple so the standby intake
         * (ha_apply_wal_catchup_entry, entry->term != 0) routes this through
         * raft_on_append_entries_request — the library does UpdateTerm/logOk/
         * (term,lsn) reconcile and on_persist_entry stamps the LEADER's term.
         * The term-0 legacy direct-apply path is retired for live entries.
         * prev_log_index == hdr->lsn - 1 (0 for lsn 1); prev_log_term is the
         * term recorded for that LSN, supplied by the caller (term index). */
        wal_catchup_entry_t entry;
        memset(&entry, 0, sizeof(entry));
        entry.lsn             = hdr->lsn;
        entry.entry_type      = (uint8_t)hdr->type;
        entry.raw_payload     = (uint8_t *)(uintptr_t)payload;
        entry.raw_payload_len = payload_len;
        entry.term            = hdr->term;
        entry.prev_log_index  = (hdr->lsn > 0) ? hdr->lsn - 1 : 0;
        entry.prev_log_term   = prev_log_term;

        wal_catchup_entry_t *ep_arr[1] = { &entry };

        int rc;
#ifdef XCE_TESTING
        if (repl_broadcast_send_hook)
            rc = repl_broadcast_send_hook(s->ep, ep_arr, 1);
        else
            rc = xce_wal_batch_req_raw(s->ep, ep_arr, 1);
#else
        rc = xce_wal_batch_req_raw(s->ep, ep_arr, 1);
#endif
        if (rc != 0) {
            WARN("repl_broadcast: send to standby %s:%u failed (rc=%d) — continuing",
                 s->host, s->port, rc);
            continue;
        }

        TRACE("repl_broadcast: sent lsn=%lu type=%u to %s:%u",
              (unsigned long)hdr->lsn, hdr->type, s->host, s->port);
        s->pending_ack_count++;
    }

    mylock_unlock(&r->lock);
    return 0;
}

/* ============================================================
 * AppendEntries send (ha-term-safety task 2.3)
 *
 * The library drives on_send_append_entries(peer_idx, prev tuple,
 * leader_commit, n_entries); the ha.c adapter sources the single entry (for
 * n_entries==1) from its own WAL and calls this to build + enqueue the
 * AppendEntries envelope to the standby at slot peer_idx. Non-blocking: reuses
 * the async xce_wal_batch_req_raw transport.
 * ============================================================ */

#ifdef XCE_TESTING
repl_ae_capture_t *repl_send_ae_capture_hook = NULL;
#endif

int repl_send_append_entries(replicator_t *r, int peer_idx,
                             uint64_t prev_log_index, uint64_t prev_log_term,
                             uint64_t leader_commit, int n_entries,
                             uint64_t entry_lsn, uint8_t entry_type,
                             uint64_t entry_term,
                             const void *payload, uint32_t payload_len)
{
    if (r == NULL || peer_idx < 0 || peer_idx >= MAX_STANDBYS)
        return -EINVAL;

#ifdef XCE_TESTING
    /* Capture seam: record the built envelope without network I/O. */
    if (repl_send_ae_capture_hook) {
        repl_ae_capture_t *cap = repl_send_ae_capture_hook;
        cap->called         = 1;
        cap->peer_idx       = peer_idx;
        cap->prev_log_index = prev_log_index;
        cap->prev_log_term  = prev_log_term;
        cap->leader_commit  = leader_commit;
        cap->n_entries      = n_entries;
        cap->entry_lsn      = (n_entries == 1) ? entry_lsn : 0;
        cap->entry_type     = (n_entries == 1) ? entry_type : 0;
        cap->entry_term     = (n_entries == 1) ? entry_term : 0;
        if (n_entries == 1 && payload && payload_len > 0) {
            uint32_t cpy = payload_len;
            if (cpy > sizeof(cap->payload))
                cpy = sizeof(cap->payload);
            memcpy(cap->payload, payload, cpy);
            cap->payload_len = cpy;
        } else {
            cap->payload_len = 0;
        }
        return 0;
    }
#endif

    mylock_lock(&r->lock);
    standby_conn_t *s = &r->standbys[peer_idx];
    repl_ep_t ep = s->ep;
    char host[XCE_MAXNAME_LEN];
    uint32_t port = s->port;
    strncpy(host, s->host, sizeof(host) - 1);
    host[sizeof(host) - 1] = '\0';
    mylock_unlock(&r->lock);

    if (ep == NULL)
        return -ENOENT; /* no live endpoint at this peer slot */

    /* Build a single AppendEntries envelope carrying the prev tuple +
     * leader_commit and, for n_entries==1, the entry sourced from the leader's
     * WAL. The carrier struct already holds the Raft term fields (task 1.1). */
    wal_catchup_entry_t entry;
    memset(&entry, 0, sizeof(entry));
    entry.prev_log_index = prev_log_index;
    entry.prev_log_term  = prev_log_term;
    /* ha-term-safety task 3.2: leader_commit is now wire-encoded on the
     * top-level container (WalCatchupEntry.leaderCommit @7) so the standby intake
     * passes it into raft_on_append_entries_request. entry is entries[0] — the
     * AppendEntries metadata holder; the encoder reads the prev tuple +
     * leader_commit from it for both the heartbeat (n_entries==0) and the
     * single-entry (n_entries==1) AE. A heartbeat carries zero entries. */
    entry.leader_commit  = leader_commit;
    if (n_entries == 1) {
        entry.lsn             = entry_lsn;
        entry.entry_type      = entry_type;
        entry.term            = entry_term;
        entry.raw_payload     = (uint8_t *)(uintptr_t)payload;
        entry.raw_payload_len = payload_len;
    }

    wal_catchup_entry_t *ep_arr[1] = { &entry };
    int rc = xce_wal_batch_req_raw(ep, ep_arr, n_entries == 1 ? 1 : 0);
    if (rc != 0) {
        WARN("repl_send_append_entries: send to standby %s:%u failed (rc=%d)",
             host, port, rc);
        return rc;
    }
    TRACE("repl_send_append_entries: peer=%d prevIdx=%lu prevTerm=%lu "
          "leaderCommit=%lu n=%d to %s:%u",
          peer_idx, (unsigned long)prev_log_index, (unsigned long)prev_log_term,
          (unsigned long)leader_commit, n_entries, host, port);
    return 0;
}

/* ============================================================
 * Sync replication — stub in P2-1; wired in P2-3
 * ============================================================ */

int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms)
{
    if (!r) return -EINVAL;

    /* Check whether any non-catching-up standby exists */
    mylock_lock(&r->lock);
    int has_standby = 0;
    for (int i = 0; i < MAX_STANDBYS; i++) {
        if (r->standbys[i].ep != NULL &&
            !atomic_load_explicit(&r->standbys[i].in_catch_up,
                                  memory_order_acquire)) {
            has_standby = 1;
            break;
        }
    }
    mylock_unlock(&r->lock);

    if (!has_standby) return -ENODEV;

    /* Compute absolute deadline for pthread_cond_timedwait (CLOCK_REALTIME) */
    struct timespec deadline;
    clock_gettime(CLOCK_REALTIME, &deadline);
    deadline.tv_sec  += (long)(timeout_ms / 1000);
    deadline.tv_nsec += (long)(timeout_ms % 1000) * 1000000L;
    if (deadline.tv_nsec >= 1000000000L) {
        deadline.tv_sec++;
        deadline.tv_nsec -= 1000000000L;
    }

    pthread_mutex_lock(&r->ack_lock);
    int rc = -ETIMEDOUT;
    while (1) {
        /* Recheck every standby after each wakeup (spurious-wakeup guard) */
        mylock_lock(&r->lock);
        for (int i = 0; i < MAX_STANDBYS; i++) {
            standby_conn_t *s = &r->standbys[i];
            if (s->ep == NULL) continue;
            if (atomic_load_explicit(&s->in_catch_up, memory_order_acquire))
                continue;
            if (atomic_load_explicit(&s->acked_lsn, memory_order_acquire)
                    >= target_lsn) {
                rc = 0;
                break;
            }
        }
        mylock_unlock(&r->lock);
        if (rc == 0) break;

        int wait_rc = pthread_cond_timedwait(&r->ack_cond, &r->ack_lock,
                                             &deadline);
        if (wait_rc == ETIMEDOUT) {
            rc = -ETIMEDOUT;
            break;
        }
    }
    pthread_mutex_unlock(&r->ack_lock);
    return rc;
}

/* ============================================================
 * Config ACK handling — call from io_worker read callback ONLY (task 8.1)
 * ============================================================ */

void repl_on_config_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_config_lsn)
{
    if (r == NULL || ep == NULL)
        return;

    mylock_lock(&r->lock);
    for (int i = 0; i < MAX_STANDBYS; i++) {
        standby_conn_t *s = &r->standbys[i];
        if (s->ep == ep) {
            atomic_store_explicit(&s->acked_config_lsn, acked_config_lsn,
                                  memory_order_release);
            break;
        }
    }
    mylock_unlock(&r->lock);

    pthread_mutex_lock(&r->ack_lock);
    pthread_cond_signal(&r->ack_cond);
    pthread_mutex_unlock(&r->ack_lock);
}

/* ============================================================
 * Majority-quorum ACK wait (task 7.3; walAck track rebased onto the
 * commit engine by ha-commit-barrier-rollout task 2.2)
 *
 * walAck:    per-LSN commit evaluation, delegated to the commit engine
 *            unconditionally (the legacy connectivity-based "ACKed
 *            anything ever" counting with self-count and the zero-active
 *            success short-circuit is retired).  Returns the engine code
 *            unmodified: 0 / -ETIMEDOUT / -XCE_ERR_NOT_READY / -EINVAL.
 * configAck: per-push arithmetic mirroring the engine's rules on the
 *            config track (barrier-rollout task 5.2, design §Config
 *            Barrier): entry_lsn carries the in-flight config push tag;
 *            count = active (non-catch-up) standbys with
 *            acked_config_lsn >= push_tag (a previous upgrade's ack no
 *            longer counts); majority = 1 (self) + count > n_peers / 2
 *            over configured members (engine n_peers, set-once at
 *            wiring); n_peers <= 1 → immediate success.  Returns 0 on
 *            quorum, -ETIMEDOUT on timeout, -EINVAL when the engine is
 *            not wired (fail-loud, mirroring the walAck branch).
 * ============================================================ */

int repl_wait_majority_ack(replicator_t *r, uint64_t entry_lsn,
                           uint32_t timeout_ms, uint8_t ack_op)
{
    if (r == NULL)
        return -EINVAL;

    if (ack_op == (uint8_t)OPCode_walAck) {
        /* LSN-0 edge: the first-ever WAL entry of a fresh cluster is LSN 0
         * (current_lsn starts at 0; written_lsn = current_lsn - 1) and is
         * structurally un-commitable — commit_submit rejects lsn == 0.
         * Documented once-per-cluster-lifetime interim behavior
         * (ha-commit-engine Implementation Notes): confirm it immediately;
         * the promotion no-op entry (ha-single-leader-role) eliminates the
         * case in practice. configAck callers also pass entry_lsn = 0 but
         * are dispatched by ack_op above, never through this guard. */
        if (entry_lsn == 0)
            return 0;

        /* Unconditional engine delegation (ha-commit-engine task 5.2,
         * design §Job-Submit Adapter). Caller holds no repl locks here, so
         * the engine's ce->lock -> r->lock order is preserved. ce == NULL
         * (engine never wired) is unreachable in production — the engine
         * is wired at cluster_init_phase2 — and commit_submit surfaces it
         * as -EINVAL rather than this adapter fabricating a quorum result. */
        return commit_submit(r->ce, entry_lsn, COMMIT_MODE_WAIT, timeout_ms);
    }

    /* ── configAck track (per-push arithmetic; barrier-rollout task 5.2) ──
     * entry_lsn carries the in-flight config push tag (repl.h: the branch
     * reuses the walAck adapter parameter — no second parameter). */

    uint64_t push_tag = entry_lsn;
    int      n_peers;

    if (r->ce == NULL) {
        /* The engine is wired at cluster_init_phase2 before any config
         * push reaches this wait; an unwired engine is a harness defect,
         * not a quorum — fail loud rather than fabricate a result
         * (mirrors the walAck branch's commit_submit -EINVAL). */
        WARN("repl_wait_majority_ack: configAck wait without a wired"
             " commit engine (push_tag=%016" PRIx64 ")", push_tag);
        return -EINVAL;
    }

    /* ce and ce->n_peers are set-once before replication traffic
     * (repl_set_commit_engine contract) — safe to read without locks. */
    n_peers = r->ce->n_peers;

    if (n_peers <= 1) {
        /* Single-member cluster: the local save is the quorum. */
        TRACE("repl_wait_majority_ack: configAck self-quorum"
              " (n_peers=%d, push_tag=%016" PRIx64 ")", n_peers, push_tag);
        return 0;
    }

    struct timespec deadline;
    clock_gettime(CLOCK_REALTIME, &deadline);
    deadline.tv_sec  += (long)(timeout_ms / 1000);
    deadline.tv_nsec += (long)(timeout_ms % 1000) * 1000000L;
    if (deadline.tv_nsec >= 1000000000L) {
        deadline.tv_sec++;
        deadline.tv_nsec -= 1000000000L;
    }

    pthread_mutex_lock(&r->ack_lock);
    int rc = -ETIMEDOUT;

    while (1) {
        mylock_lock(&r->lock);

        int covered = 0;

        for (int i = 0; i < MAX_STANDBYS; i++) {
            standby_conn_t *s = &r->standbys[i];
            if (s->ep == NULL)
                continue;
            if (atomic_load_explicit(&s->in_catch_up, memory_order_acquire))
                continue;
            if (atomic_load_explicit(&s->acked_config_lsn,
                                     memory_order_acquire) >= push_tag)
                covered++;
        }

        mylock_unlock(&r->lock);

        /* Majority over configured members, self counted: the primary's
         * own save plus standbys covering the in-flight push tag.  No
         * n_active shortcut — a multi-member cluster with zero covering
         * acks must time out, never fabricate success (Req 4.2). */
        if (1 + covered > n_peers / 2) {
            rc = 0;
            break;
        }

        int wait_rc = pthread_cond_timedwait(&r->ack_cond, &r->ack_lock,
                                             &deadline);
        if (wait_rc == ETIMEDOUT) {
            rc = -ETIMEDOUT;
            break;
        }
    }

    pthread_mutex_unlock(&r->ack_lock);

    if (rc == 0)
        INFO("repl_wait_majority_ack: quorum reached (ack_op=%u)", ack_op);
    else
        WARN("repl_wait_majority_ack: timeout waiting for majority quorum"
             " (ack_op=%u, timeout_ms=%u)", ack_op, timeout_ms);

    return rc;
}

/* ============================================================
 * WAL trim guard
 * ============================================================ */

uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn,
                        uint32_t snap_interval_s)
{
    (void)snap_interval_s;

    if (r == NULL) {
        return checkpoint_lsn;
    }

    uint64_t min_lsn = checkpoint_lsn;

    mylock_lock(&r->lock);

    for (int i = 0; i < MAX_STANDBYS; i++) {
        const standby_conn_t *s = &r->standbys[i];

        if (s->ep != NULL) {
            /* Active standby: guard at its last acked LSN */
            uint64_t acked = atomic_load(&s->acked_lsn);
            if (acked < min_lsn) {
                min_lsn = acked;
            }
        } else if (s->last_disconnected_lsn != 0) {
            /* Recently disconnected slot still holding a guard */
            if (s->last_disconnected_lsn < min_lsn) {
                min_lsn = s->last_disconnected_lsn;
            }
        }
    }

    mylock_unlock(&r->lock);

    return min_lsn;
}

/* ============================================================
 * Catch-up state management (P2-5)
 * ============================================================ */

void repl_start_catchup(replicator_t *r, repl_ep_t ep, uint64_t join_applied_lsn)
{
    int changed = 0;

    mylock_lock(&r->lock);
    for (int i = 0; i < MAX_STANDBYS; i++) {
        if (r->standbys[i].ep == ep) {
            r->standbys[i].join_applied_lsn = join_applied_lsn;
            atomic_store_explicit(&r->standbys[i].in_catch_up, 1,
                                  memory_order_release);
            changed = 1;
            break;
        }
    }
    mylock_unlock(&r->lock);

    /* Catch-up entry excludes the standby from the evaluation count:
     * re-evaluate pending commits (after r->lock release — engine lock
     * order; ha-commit-engine task 5.2, Req 4.2). */
    if (changed && r->ce != NULL) {
        commit_on_membership_change(r->ce);
    }
}

void repl_complete_catchup(replicator_t *r, repl_ep_t ep)
{
    int changed = 0;

    mylock_lock(&r->lock);
    for (int i = 0; i < MAX_STANDBYS; i++) {
        if (r->standbys[i].ep == ep) {
            atomic_store_explicit(&r->standbys[i].in_catch_up, 0,
                                  memory_order_release);
            /* bug-1479: reset the eviction counter on re-entry to live
             * broadcast -- otherwise a stale pending_ack_count carried over
             * from before catch-up could immediately re-trip
             * repl_broadcast's MAX_PENDING_ACKS eviction on the very next
             * send, undoing the catch-up this function just completed. */
            r->standbys[i].pending_ack_count = 0;
            changed = 1;
            break;
        }
    }
    mylock_unlock(&r->lock);

    /* Standby rejoins the evaluation count: its watermark may newly
     * satisfy a pending majority — re-evaluate (after r->lock release —
     * engine lock order; ha-commit-engine task 5.2, Req 4.2). */
    if (changed && r->ce != NULL) {
        commit_on_membership_change(r->ce);
    }

    /* Broadcast so any blocked repl_wait_ack() re-checks the ACK condition */
    pthread_mutex_lock(&r->ack_lock);
    pthread_cond_broadcast(&r->ack_cond);
    pthread_mutex_unlock(&r->ack_lock);
}

int repl_get_catchup_join_lsn(replicator_t *r, repl_ep_t ep,
                              uint64_t *join_applied_lsn_out)
{
    if (r == NULL || join_applied_lsn_out == NULL)
        return 0;

    int found = 0;
    mylock_lock(&r->lock);
    for (int i = 0; i < MAX_STANDBYS; i++) {
        if (r->standbys[i].ep == ep) {
            if (atomic_load_explicit(&r->standbys[i].in_catch_up,
                                     memory_order_acquire)) {
                *join_applied_lsn_out = r->standbys[i].join_applied_lsn;
                found = 1;
            }
            break;
        }
    }
    mylock_unlock(&r->lock);
    return found;
}

int repl_needs_snapshot(replicator_t *r, uint64_t standby_lsn,
                        uint64_t oldest_wal_lsn)
{
    (void)r;
    /* No WAL at all: empty standby can start from LSN 0 without a snapshot */
    if (oldest_wal_lsn == 0) return 0;
    /* Snapshot only when the standby's NEXT-needed entry (standby_lsn + 1) has
     * been GC'd below the oldest surviving WAL segment — i.e. it can no longer be
     * served by streaming.  An empty standby (standby_lsn=0) whose next-needed
     * entry IS the oldest retained entry (e.g. lsn 1 post reserve-lsn-0) is
     * streamable: 0 + 1 < 1 is false → no snapshot.  Mirrors the boundary in
     * ha_evaluate_repl_action (ha.c). */
    return (standby_lsn + 1 < oldest_wal_lsn) ? 1 : 0;
}

/* ============================================================
 * Write-admission quorum accounting (ha-commit-engine task 4)
 * ============================================================ */

/* repl_active_standby_count — registry-truth count of standbys able to
 * receive new log entries: slots with ep != NULL && !in_catch_up, evaluated
 * under r->lock. NULL registry counts zero. Consumed by ha_has_write_quorum
 * (ha.c) for write admission. Takes ONLY r->lock — never the commit-engine
 * mutex (lock-ordering rule: gate path holds no locks when calling in).
 * (ha-commit-engine Req 3.1, 3.2, 3.3) */
/* repl_set_commit_engine — wire the commit engine into the replicator
 * (ha-commit-engine task 5.2). Once set, repl_on_ack feeds commit_on_ack,
 * the membership mutators feed commit_on_membership_change, and the walAck
 * branch of repl_wait_majority_ack delegates to commit_submit. Called once
 * at cluster init before replication traffic (task 5.3); NULL un-wires. */
void repl_set_commit_engine(replicator_t *r, struct commit_engine *ce)
{
    if (r == NULL) {
        return;
    }
    r->ce = ce;
}

int repl_active_standby_count(replicator_t *r)
{
    int i;
    int count = 0;

    if (!r)
        return 0;

    mylock_lock(&r->lock);
    for (i = 0; i < MAX_STANDBYS; i++) {
        if (r->standbys[i].ep != NULL &&
            !atomic_load_explicit(&r->standbys[i].in_catch_up,
                                  memory_order_acquire))
            count++;
    }
    mylock_unlock(&r->lock);

    return count;
}