myxce/src/include/repl.h

#pragma once
/*
 * repl.h — P2-1: Replicator structures and public API.
 * Fan-out TCP replication via existing iomgr/SSL endpoint pipeline.
 * ACKs processed in io_worker read callback (NOT vtaskmgr) to prevent deadlock.
 */
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <time.h>
#include "config.h"
#include "mylock.h"
#include "wal_types.h"  /* wal_entry_hdr_t — P2-4: use wal_types.h to break circular include */

/* Max queued-but-unacknowledged sends per standby before eviction to catch-up */
#define MAX_PENDING_ACKS 16

/* Endpoint is an opaque pointer — full type in netio.h (avoid heavy include) */
typedef void* repl_ep_t;

/* Commit engine (controller/include/commit.h) — forward-declared so repl.h
 * keeps zero controller-header dependencies (ha-commit-engine task 5). */
struct commit_engine;

/* Per-standby connection slot */
typedef struct {
    repl_ep_t          ep;                /* NULL = slot empty */
    _Atomic uint64_t   acked_lsn;        /* Last WAL LSN ACKed (walAck) */
    _Atomic uint64_t   acked_config_lsn; /* Last config LSN ACKed (upgradeConfigAck) */
    _Atomic int        in_catch_up;      /* Non-zero while catching up */
    uint64_t           join_applied_lsn; /* LSN standby reported on reconnect */
    char               host[XCE_MAXNAME_LEN];
    uint32_t           port;
    uint64_t           last_disconnected_lsn;
    time_t             last_disconnect_time;
    uint32_t           pending_ack_count; /* Sends not yet ACKed; evict if > MAX_PENDING_ACKS */
} standby_conn_t;

/* Replicator — embedded in persist_t or cluster_t (P2-4).
 * Named tag replicator_s matches the forward declaration in persist.h. */
typedef struct replicator_s {
    standby_conn_t  standbys[MAX_STANDBYS];
    int             nstandbys;          /* Active standby count */
    mylock_t        lock;               /* Protects standbys[], nstandbys */
    pthread_cond_t  ack_cond;           /* Signaled when any ACK arrives */
    pthread_mutex_t ack_lock;           /* Guards ack_cond */
    /* Commit engine feed (ha-commit-engine task 5.2). NULL = not wired
     * (legacy counting contexts). Set once via repl_set_commit_engine at
     * cluster init, before replication traffic — read without r->lock.
     * Lock ordering: engine paths take ce->lock then r->lock, so repl.c
     * must NEVER call into the engine while holding r->lock. */
    struct commit_engine *ce;
} replicator_t;

/* Lifecycle */
int  repl_init(replicator_t *r);
void repl_fini(replicator_t *r);

/* Standby management */
int  repl_add_standby(replicator_t *r, repl_ep_t ep, const char *host, uint32_t port);
void repl_remove_standby(replicator_t *r, repl_ep_t ep);
void repl_expire_disconnect_slots(replicator_t *r, time_t now_s, uint32_t snap_interval_s);

/* Write-admission quorum accounting (ha-commit-engine Req 3.1-3.3).
 * Counts only 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). */
int  repl_active_standby_count(replicator_t *r);

/* Replication send (stub in P2-1; wired in P2-4).
 *
 * ha-term-safety task 3.4 (Req 1.3/1.4): the LIVE primary append replicator.
 * Carries the leader's per-entry Raft term (`hdr->term`) and the AppendEntries
 * prev tuple (`prev_log_index = hdr->lsn - 1`, `prev_log_term` supplied by the
 * caller from its term index) on every broadcast, so the standby intake routes
 * the entry through `raft_on_append_entries_request` and persists the LEADER's
 * term (not a local substitute or 0). `prev_log_term` is the term recorded for
 * `hdr->lsn - 1` (0 when `hdr->lsn == 1`, i.e. no predecessor). The term-0
 * legacy direct-apply receive path is thereby retired for live entries. */
int  repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
                    const void *payload, uint32_t payload_len,
                    uint64_t prev_log_term);

/* AppendEntries send (ha-term-safety task 2.3, Req 1.3/3.1).
 *
 * Build and enqueue ONE AppendEntries envelope to the standby at slot
 * `peer_idx` (== the library's raft peer index, which maps 1:1 to the repl
 * standby registry slot — the same index used by match_index/next_index).
 * The envelope carries the AppendEntries prev tuple (`prev_log_index`,
 * `prev_log_term`) + `leader_commit`; when `n_entries == 1` it also carries the
 * single entry at LSN `prev_log_index + 1` whose `entry_term` is the term the
 * leader recorded for that LSN (== persist_entry_term) and whose
 * `entry_type`/`payload` are read from the leader's own WAL. `n_entries == 0`
 * is a heartbeat (zero entries) carrying only the prev tuple + leader_commit.
 *
 * Non-blocking: enqueues via the existing async repl transport
 * (xce_wal_batch_req_raw), never blocks. Returns 0 on enqueue, or a negative
 * errno (e.g. -ENOENT when peer_idx has no live endpoint). */
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);

/* WAL ACK handling — call from io_worker read callback ONLY */
void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn);

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

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

/* Majority-quorum ACK wait (task 7.3; ha-commit-engine task 5 adapter;
 * configAck track hardened by ha-commit-barrier-rollout task 5.2).
 * entry_lsn: walAck — the WAL entry the caller waits on, delegated to the
 * commit engine for per-LSN evaluation; configAck — the in-flight config
 * push tag, counted as acked_config_lsn >= tag with majority
 * 1 (self) + count > n_peers / 2 over configured members (no second
 * parameter — the configAck branch reuses this one). */
int  repl_wait_majority_ack(replicator_t *r, uint64_t entry_lsn,
                            uint32_t timeout_ms, uint8_t ack_op);

/* Wire the commit engine into the replicator so ACK and membership events
 * are fed to it (ha-commit-engine task 5.2; NULL = not wired). */
void repl_set_commit_engine(replicator_t *r, struct commit_engine *ce);

/* WAL trim guard */
uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn, uint32_t snap_interval_s);

/* Catch-up state management */

/* Mark standby as entering catch-up mode; stores join_applied_lsn */
void repl_start_catchup(replicator_t *r, repl_ep_t ep, uint64_t join_applied_lsn);

/* Mark standby catch-up complete (called on OP_replCatchUpDone receipt) */
void repl_complete_catchup(replicator_t *r, repl_ep_t ep);

/* bug-1479: look up whether the given standby is currently in catch-up and,
 * if so, its corrected resume point (join_applied_lsn). Returns 1 and sets
 * *join_applied_lsn_out when in_catch_up is set; returns 0 (leaves
 * *join_applied_lsn_out untouched) otherwise. Lets a caller (the heartbeat-
 * driven replication decision) use the corrected position instead of a
 * standby's own self-reported LSN, which is unreliable for a standby stuck
 * rejecting every AppendEntries due to a log conflict. */
int repl_get_catchup_join_lsn(replicator_t *r, repl_ep_t ep,
                              uint64_t *join_applied_lsn_out);

/* Returns 1 if standby needs a full snapshot transfer (its LSN predates oldest WAL).
 * oldest_wal_lsn == 0 means no WAL exists — no snapshot needed for LSN 0 standbys. */
int repl_needs_snapshot(replicator_t *r, uint64_t standby_lsn, uint64_t oldest_wal_lsn);

#ifdef XCE_TESTING
/* AppendEntries send capture seam (ha-term-safety task 2.3).
 *
 * When `repl_send_ae_capture_hook` is non-NULL, repl_send_append_entries
 * records the FULL built envelope into this struct instead of enqueuing it on a
 * real endpoint — so a unit test asserts the prev tuple / leader_commit /
 * n_entries / entry term + payload without network I/O. The payload bytes are
 * copied into a fixed buffer (capped at sizeof(payload)). */
typedef struct {
    int      called;
    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;
    uint32_t payload_len;
    uint8_t  payload[256];
} repl_ae_capture_t;

extern repl_ae_capture_t *repl_send_ae_capture_hook;
#endif