forked from lei.chen/myxce
Rongsong Shen
5a2fae1d37
configAck branch: per-push majority (acked_config_lsn >= push_tag, 1+count > n_peers/2 over r->ce->n_peers, n_peers<=1 immediate, ce==NULL -EINVAL); legacy >0 counting and n_active==0 shortcut dead. push_config_to_standbys returns the barrier result; upgrade failure -> send_upgrade_error (rc forwarded), no success ACK, no restart, committed_config_tag success-only. Remediation r1 (two CRITICAL review findings): - F1: wire production config-ACK ingestion — replconfigpushreplymsg decode branch in repl_ack_decode_and_dispatch -> repl_on_config_ack on the promoted-vote ep; + defensive cluster.c OPCode_reply case. Closes the unlanded half of ha-config-bootstrap 8.1; without this the barrier would brick every multi-member config upgrade. No schema change. - F2: keep the barrier off the iomgr thread — export push_config_transport_only (non-blocking) for the 5 ha.c + join.c streaming sites; barrier path now vtask-only (do_upgrade_config). Wire-path test drives real capnp ACK bytes through the production decode branch. 30/30 rollout, 6/6 repl_majority, 46/46 config_repl; census 0 FAIL; pre-existing fixture leak only (less than baseline).
110 lines
5.1 KiB
C
110 lines
5.1 KiB
C
#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) */
|
|
int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
|
|
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);
|