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