wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

feat(ha-wal-replication): HA→replicator standby registration bridge (tasks 2.1-2.3)

Browse source 
Register standbys with replicator on connect (election promotion +
late-joiner), deregister on disconnect (vote err handler + standby err
handler). repl_add_standby now idempotent (duplicate ep returns
existing slot).

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-06-10 08:35:02 +00:00
parent 18e351ee1b
commit 81dbc4d7b4
4 changed files with 542 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

124
.kiro/specs/ha-wal-replication/tasks.md Normal file
View file

@ -0,0 +1,124 @@
# Implementation Plan: HA WAL Replication
## Task Overview
5 requirements mapped to 4 implementation groups + 1 validation group. Bridging `repl.c` (replicator) and `ha.c`/`config_repl.c` (HA connection layer) so WAL entries transit from primary to standbys and are majority-committed before the caller is acknowledged.
---
- [ ] 1. repl_broadcast TCP send
- [x] 1.1 Fill in repl_broadcast send path
- Remove `(void)payload; (void)payload_len;` and the `pending_ack_count++` counter.
- For each non-catch-up standby (`s->ep != NULL, !in_catch_up`), build a single-entry `wal_catchup_entry_t` with `lsn=hdr->lsn`, `entry_type=hdr->type`, `raw_payload=payload`, `raw_payload_len=payload_len`.
- Enqueue via `xce_wal_batch_req_raw(s->ep, &entry, 1)`.
- If `xce_wal_batch_req_raw` fails, log WARN and continue to next standby.
- Pending-ack-count eviction guard (line 217-225) stays; evicted standbys skip the send.
- Compilation succeeds; `/xce-build controller` green.
- _Requirements: 2.1, 2.2, 2.3, 2.4_
- _Boundary: Replicator (repl.c)_
- [x] 1.2 Verify no standbys → immediate return
- Confirm existing guard: when `r->standbys[]` is all-NULL, loop iterates zero times, returns 0.
- Add unit test: alloc replicator with zero standbys, call `repl_broadcast`, verify returns 0 without crashing or sending.
- _Requirements: 2.4_
- _Boundary: Replicator (repl.c)_
---
- [ ] 2. HA→replicator standby registration bridge
- [x] 2.1 Register standbys on connect
- In `raft_cb_become_leader` (ha.c), after promoting vote eps to `ha->standby_endpoints`: for each promoted ep, call `repl_add_standby(&c->persist.repl, ep, peer_host, peer_port)`.
- In `on_standby_connected` (ha.c), after ctrlhello is sent: call `repl_add_standby` with the late-joiner's endpoint, host, and port.
- Host/port obtained from `c->myconf->peers_cfg.candidates[i]`.
- Log DEBUG3 on successful registration.
- _Requirements: 1.1_
- _Boundary: HA connection layer (ha.c, config_repl.c)_
- [x] 2.2 Deregister standbys on disconnect
- In `vote_ep_err_handler` (ha.c): before or after removing ep from `ha->standby_endpoints`, call `repl_remove_standby(&c->persist.repl, ep)`.
- In `standby_ep_err_handler_primary` (ha.c): call `repl_remove_standby` before endpoint cleanup.
- `repl_remove_standby` is NULL-safe (returns early if ep NULL).
- _Requirements: 1.2_
- _Boundary: HA connection layer (ha.c)_
- [x] 2.3 Idempotent re-registration
- `repl_add_standby` already handles duplicate ep: it finds the existing slot by ep pointer match, resets fields.
- Unit test: call `repl_add_standby` twice with same ep, verify no double-count and no crash.
- _Requirements: 1.3_
- _Boundary: Replicator (repl.c)_
---
- [ ] 3. Standby WAL ACK reply
- [ ] 3.1 Send WalAckReply after WAL entry applied
- In `ha_apply_wal_catchup_entry` (ha.c), after `ha_apply_single_entry` succeeds for an entry: construct a `WalAckReply { term=ha->term, lsn=ent->lsn, success=1 }` and send via `xce_reply` to the endpoint that delivered the entry.
- The endpoint is not directly available in `ha_apply_wal_catchup_entry` — it's called from `cluster.c:handle_wal_catchup_entry` which has the ep. Pass `ep` as a new parameter: `ha_apply_wal_catchup_entry(ha, p, entry, ep)`.
- _Requirements: 3.1, 3.2_
- _Boundary: Standby dispatch (cluster.c, ha.c)_
- [ ] 3.2 Skip ACK on apply failure
- In `ha_apply_single_entry` (ha.c): return distinct error codes for skip-vs-crash.
- In `ha_apply_wal_catchup_entry`: if `ha_apply_single_entry` returns non-zero, skip the ACK for that entry (already logged as ERROR).
- _Requirements: 3.3_
- _Boundary: Standby dispatch (ha.c)_
- [ ] 3.3 Standby role guard
- The existing `ha_dispatch_gate` already allows `OPCode_walCatchupEntry` in `CTRL_STANDBY`, `CTRL_ELECTING`, and `CTRL_INIT`.
- Verify with unit test: standby applies WAL entry while in `CTRL_STANDBY`, `CTRL_ELECTING`, `CTRL_INIT` — all succeed.
- _Requirements: 3.4_
- _Boundary: Standby dispatch (cluster.c, ha.c)_
---
- [ ] 4. Primary ACK handler → advance committed_lsn
- [ ] 4.1 Handle incoming WalAckReply on primary
- In `repl_ack_decode_and_dispatch` (ha.c:2506), add case for `RequestReply_data_walackreplymsg`.
- Extract `lsn` and `success` from `WalAckReply`. If `success`, call `repl_on_ack(&c->persist.repl, ep, lsn)`.
- `repl_on_ack` (repl.c:158) already updates `acked_lsn` and signals `ack_cond` — no changes needed there.
- _Requirements: 4.3_
- _Boundary: Primary heartbeat handler (ha.c)_
- [ ] 4.2 Verify majority-ack advances committed_lsn
- `persist_wal_sync_append` already calls `repl_wait_majority_ack` and advances `committed_lsn` on success.
- Integration test: start 3-node cluster, submit a job, verify `committed_lsn > 0` on primary after submit.
- _Requirements: 4.1, 4.2, 4.4_
- _Boundary: Persistence (persist.c, repl.c)_
---
- [ ] 5. Validation: tests and integration
- [ ] 5.1 Unit test: repl_broadcast sends to registered standby
- Alloc replicator, register a mock endpoint via `repl_add_standby`.
- Call `repl_broadcast` with test header + payload.
- Verify `xce_wal_batch_req_raw` was invoked (via test hook pattern).
- Verify payload bytes round-trip through encode→decode.
- _Requirements: 2.1, 5.2_
- _Boundary: Test (testsuite/test_ha_wal_repl.c)_
- [ ] 5.2 Unit test: ACK round-trip
- Standby side: encode WalAckReply, decode on primary side, call `repl_on_ack`.
- Verify `acked_lsn` updated, `ack_cond` signaled, `repl_wait_majority_ack` unblocks.
- _Requirements: 4.3, 5.3_
- _Boundary: Test (testsuite/test_ha_wal_repl.c)_
- [ ] 5.3 Unit test: registration lifecycle
- Register, verify `nstandbys==1` and `standbys[0].ep == ep`.
- Register duplicate, verify `nstandbys==1` (idempotent).
- Deregister, verify slot freed (`ep==NULL`, `nstandbys==0`).
- _Requirements: 1.1, 1.2, 1.3_
- _Boundary: Test (testsuite/test_ha_wal_repl.c)_
- [ ] 5.4 Integration test: 3-node cluster WAL replication
- Deploy via `/xce-deploy-ha`; verify primary elected.
- Submit job via `jobsub`; verify job appears in `joblist` on primary AND on all standbys.
- Kill primary; verify promoted standby has the job (failover durability).
- Verify `committed_lsn` advanced on all nodes.
- _Requirements: 4.1, 5.5_
- _Boundary: Integration (deploy-ha)_
- [ ] 5.5 ASAN and regression
- Full controller unit suite green under ASAN.
- No new ASAN leaks or errors vs baseline.
- `/xce-build controller` green.
- _Requirements: 5.5_
- _Boundary: Build (meson)_

154
src/controller/ha.c
View file

@ -510,6 +510,14 @@ static void raft_cb_become_leader(void *ctx, uint64_t term)
/* Reset ack counter for this committing round */
atomic_store(&ha->repl_ack_count, 0);
/* Req 2.1 (ha-repl-hardening): re-initialize per-standby convergence
* tracking for THIS committing round. Stale converged flags from a
* previous term (demote → re-elect cycle) must not satisfy the
* majority-convergence gate before standbys re-report their state. */
memset(ha->standby_reported_tag, 0, sizeof(ha->standby_reported_tag));
memset(ha->standby_reported_lsn, 0, sizeof(ha->standby_reported_lsn));
memset(ha->standby_converged, 0, sizeof(ha->standby_converged));
/* Promote surviving vote endpoints to standby replication channels.
* Iterate ha->elect_ctx->votes[i].ep and add non-NULL entries to
* ha->standby_endpoints so push_config and ctrlhello reach them. */
@ -522,6 +530,20 @@ static void raft_cb_become_leader(void *ctx, uint64_t term)
if (ectx->votes[i].ep != NULL) {
zarray_store(ha->standby_endpoints, ectx->votes[i].ep);
INFO("HA: promoted vote ep[%d] → standby channel", i);
/* Task 2.1: register promoted ep with WAL replicator.
* Req 1.1: primary must register standby with repl_add_standby
* when establishing the standby control channel.
* peer->hostname / peer->port from the vote entry's peer pointer. */
if (c && ectx->votes[i].peer != NULL) {
int rc = repl_add_standby(&c->persist.repl,
ectx->votes[i].ep,
ectx->votes[i].peer->hostname,
(uint32_t)ectx->votes[i].peer->port);
if (rc >= 0)
TRACE("HA: repl registered standby ep[%d] (slot %d)", i, rc);
else
WARN("HA: repl_add_standby ep[%d] failed: %d", i, rc);
}
}
}
pthread_mutex_unlock(&ha->lock);
@ -970,6 +992,9 @@ static void ha_advance_applied_lsn(ha_state_t *ha, uint64_t new_lsn)
}
/* ha_apply_single_entry: apply one decoded wal_catchup_entry_t sub-entry.
* Payload bytes (raw_payload/raw_payload_len, extracted by xce_decode)
* are passed through to apply_wal_entry so entity state is restored —
* not just the LSN chain (task 7.3).
* Returns 0 on success, non-zero on error (caller logs and skips). */
static int ha_apply_single_entry(persist_t *p, wal_catchup_entry_t *e)
{
@ -977,10 +1002,10 @@ static int ha_apply_single_entry(persist_t *p, wal_catchup_entry_t *e)
memset(&hdr, 0, sizeof(hdr));
hdr.lsn = e->lsn;
hdr.type = (uint32_t)e->entry_type;
hdr.payload_len = 0;
hdr.payload_len = (uint32_t)e->raw_payload_len;
hdr.crc = 0;
hdr.prev_crc = 0;
return apply_wal_entry(p, &hdr, NULL);
return apply_wal_entry(p, &hdr, e->raw_payload);
}
int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
@ -1051,7 +1076,9 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p,
* 3. Load the snapshot via persist_load_snapshot().
* 4. Set ha->applied_lsn = p->checkpoint_lsn (in-memory).
*
* On mid-transfer reconnect callers discard snap.part externally (Req 4.4).
* Mid-transfer reconnect (Req 4.4): the primary restarts every transfer at
* chunk 0, so chunk_index == 0 discards any stale snap.part left by an
* interrupted transfer before appending — no cross-chunk state needed.
* Returns 0 on success, negative errno on error.
* Requirements: 4.2, 4.3, 4.4 (ha-repl-hardening spec)
*/
@ -1067,6 +1094,12 @@ int ha_handle_snapshot_chunk(ha_state_t *ha, persist_t *p,
char part_path[4096 + 16];
snprintf(part_path, sizeof(part_path), "%s/snap.part", snap_dir);
/* Req 4.4: a transfer always restarts from chunk 0 — discard any stale
* partial file from an interrupted previous transfer. */
if (chunk->chunk_index == 0 && unlink(part_path) != 0 && errno != ENOENT)
WARN("ha_handle_snapshot_chunk: unlink stale snap.part failed: %s",
strerror(errno));
/* Append chunk payload to snap.part */
if (payload_bytes && payload_len > 0) {
FILE *f = fopen(part_path, "ab");
@ -1258,13 +1291,19 @@ void ha_check_majority_convergence(ha_state_t *ha, cluster_t *c)
n_converged++;
}
/* Majority threshold: all standbys must converge (n_peers - 1).
* Single-node (n_peers=1): threshold=0 — satisfied immediately. */
if (n_converged < n_peers - 1)
/* Majority threshold (Req 2.3): primary + converged standbys must form
* a majority of the cluster — ceil(n_peers/2) - 1 standbys, integer form
* n_peers / 2 (n=3 → 1, n=5 → 2). Safe: committed entries are already
* on a majority (per-entry ack) and safe_committed_lsn derives from a
* majority of voters; a lagging minority standby keeps catching up via
* the heartbeat-driven decision loop after promotion.
* Single-node (n_peers<=1): threshold=0 — satisfied immediately. */
int threshold = n_peers / 2;
if (n_converged < threshold)
return;
INFO("HA: majority converged (%d/%d standbys) → CTRL_PRIMARY",
n_converged, n_peers);
INFO("HA: majority converged (%d/%d standbys, threshold %d) → CTRL_PRIMARY",
n_converged, n_peers - 1, threshold);
if (c) {
/* Full promotion: increments term, sets role=PRIMARY, registers hooks,
@ -1355,6 +1394,14 @@ int ha_dispatch_repl_action_for_standby(ha_state_t *ha,
if (c && c->persist.snap_path[0]) {
extern int ha_send_snapshot_to_standby(endpoint_t *, const char *,
uint64_t);
/* Req 4.1: create a FRESH checkpoint so the transferred snapshot
* reflects current state (also trims WAL + rotates old snaps).
* -EBUSY means another checkpoint is in flight — fall back to
* the existing checkpoint_lsn rather than skipping the cycle. */
int ckpt_rc = persist_checkpoint(&c->persist);
if (ckpt_rc != 0 && ckpt_rc != -EBUSY)
WARN("HA repl: persist_checkpoint before snapshot send "
"failed rc=%d — sending existing checkpoint", ckpt_rc);
ha_send_snapshot_to_standby(ep, c->persist.snap_path,
c->persist.checkpoint_lsn);
}
@ -1419,17 +1466,35 @@ int ha_dispatch_gate(ha_state_t *ha, int opcode)
return 0;
case CTRL_INIT:
/* Allow election probes, config probes, AND heartbeats so the primary's
* periodic heartbeat can reach the handler and post heartbeat_seen,
* triggering the CTRL_INIT → CTRL_STANDBY transition (Req 1.1). */
/* Allow election RPCs AND primary→standby replication messages.
* During the startup stagger a primary may already connect and push
* config — accepting these ensures the node doesn't miss the initial
* bootstrap and can transition to CTRL_STANDBY immediately. */
if (opcode == OPCode_requestVote ||
opcode == OPCode_probeConfigTag ||
opcode == OPCode_heartbeat)
opcode == OPCode_heartbeat ||
opcode == OPCode_hello ||
opcode == OPCode_nop ||
opcode == OPCode_replConfigPush ||
opcode == OPCode_walCatchupEntry ||
opcode == OPCode_snapshotTransfer)
return 0;
return -XCE_ERR_NOT_READY;
case CTRL_ELECTING:
if (opcode == OPCode_requestVote || opcode == OPCode_probeConfigTag)
/* Allow election RPCs AND primary→standby replication messages.
* A node campaigning for leadership must still accept config, WAL,
* and heartbeats from the current primary — if it wins the election
* it needs current state; if it loses, it becomes STANDBY and must
* resume replication without having missed pushes. Only job/client
* ops are rejected (we're not ready to serve). */
if (opcode == OPCode_requestVote ||
opcode == OPCode_probeConfigTag ||
opcode == OPCode_hello ||
opcode == OPCode_nop ||
opcode == OPCode_replConfigPush ||
opcode == OPCode_walCatchupEntry ||
opcode == OPCode_snapshotTransfer)
return 0;
return -XCE_ERR_NOT_READY;
@ -1789,7 +1854,13 @@ static int ha_heartbeat_is_fresh(ha_state_t *ha)
* 6. grant vote, record voted_for, persist (Req 9.1)
*
* Returns 1 = granted, 0 = denied, <0 = error.
*
* TEST-ONLY (XCE_TESTING): production vote dispatch goes through
* cluster.c → raft_on_vote_request. This function is the legacy
* entry point used by test_ha_vote_gate.c and test_ha_election.c
* to exercise role-gate logic in isolation without a full raft state.
*/
#ifdef XCE_TESTING
int handle_request_vote(ha_state_t *ha,
uint64_t term,
const char *candidate_id,
@ -1887,6 +1958,7 @@ int handle_request_vote(ha_state_t *ha,
return 1;
}
#endif /* XCE_TESTING */
/* ── Test-only hook shims (XCE_TESTING only) ──────────────────────────────
* Type definitions (election_timeout_ctx_t, committing_timeout_ctx_t),
@ -2371,6 +2443,9 @@ static void vote_reply_dispatch(ha_state_t *ha, ha_elect_ctx_t *ctx, int idx,
* to vote_reply_dispatch. Factored from vote_reply_msg_handler so that
* peer_ctrl_msg_handler can invoke the same logic without code duplication.
* Packet ownership: caller passes ep->fsm_in.pkt; this function frees it.
* ep->fsm_in.pkt is cleared HERE, before dispatch — vote_reply_dispatch can
* reach raft_cb_become_follower → endpoint_fini(ep) (loser path), which
* would otherwise double-free the packet via the stale fsm_in.pkt pointer.
*/
static int vote_reply_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
{
@ -2380,6 +2455,9 @@ static int vote_reply_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
io_packet_t *pkt = ep->fsm_in.pkt;
int ok = 0;
ep->fsm_in.pkt = NULL; /* take ownership (same pattern as
* repl_ack_decode_and_dispatch) */
uint64_t vr_term = 0;
int vr_granted = 0;
uint64_t vr_tag = 0;
@ -2966,6 +3044,12 @@ static int vote_ep_err_handler(void *ep_raw)
}
pthread_mutex_unlock(&ha->lock);
}
/* Task 2.2: deregister from WAL replicator.
* Req 1.2: when a standby disconnects, the HA layer shall deregister
* the endpoint from the replicator via repl_remove_standby.
* vctx->c may be NULL in test mode — guard accordingly. */
if (vctx->c)
repl_remove_standby(&vctx->c->persist.repl, ep);
} else {
/* Election phase: count as denied vote */
handle_request_vote_reply(ha, vctx->ctx, vctx->peer_idx,
@ -3455,6 +3539,22 @@ void ha_arm_heartbeat_timeout_vtask(ha_state_t *ha, cluster_t *c)
}
}
/* ha_discover_primary: if CTRL_ELECTING, step down to STANDBY on receiving
* a valid primary message. A campaigning node that sees a live primary must
* defer — continuing the election risks split-brain (bug-353). */
void ha_discover_primary(ha_state_t *ha, cluster_t *c)
{
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
if (role != CTRL_ELECTING)
return;
uint64_t term = atomic_load_explicit(&ha->term, memory_order_acquire);
INFO("HA: discovered live primary while ELECTING (term=%" PRIu64 ") — stepping down",
term);
raft_cb_become_follower(ha, term, -1);
}
void ha_on_heartbeat_received(ha_state_t *ha)
{
if (!ha) return;
@ -3587,6 +3687,25 @@ static void on_standby_connected(endpoint_t *ep)
INFO("HA: outbound standby connection to peer[%d] established, ctrlhello sent",
sctx->peer_idx);
/* Task 2.1: register late-joiner with WAL replicator.
* Req 1.1: primary must register standby with repl_add_standby when
* establishing the standby control channel (late-joiner outbound connect).
* Host/port from candidates[peer_idx] — the peer we connected to. */
if (c && c->myconf &&
sctx->peer_idx >= 0 &&
sctx->peer_idx < c->myconf->peers_cfg.candidates_count) {
const xce_candidate_t *peer =
&c->myconf->peers_cfg.candidates[sctx->peer_idx];
int rc = repl_add_standby(&c->persist.repl, ep,
peer->hostname, (uint32_t)peer->port);
if (rc >= 0)
TRACE("HA: repl registered late-joiner standby peer[%d] (slot %d)",
sctx->peer_idx, rc);
else
WARN("HA: repl_add_standby late-joiner peer[%d] failed: %d",
sctx->peer_idx, rc);
}
/* Set state-aware dispatcher for subsequent messages on this endpoint */
ep->msg_handler = peer_ctrl_msg_handler;
}
@ -3603,6 +3722,8 @@ static int standby_ep_err_handler_primary(void *ep_raw)
if (!sctx) return 0;
ha_state_t *ha = sctx->ha;
cluster_t *c = sctx->c;
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
for (int i = 0; i < ha->standby_endpoints->count; i++) {
@ -3613,6 +3734,13 @@ static int standby_ep_err_handler_primary(void *ep_raw)
}
pthread_mutex_unlock(&ha->lock);
}
/* Task 2.2: deregister late-joiner standby from WAL replicator before
* freeing the context. Req 1.2: RDHUP or connection error triggers
* repl_remove_standby so the replicator stops sending WAL entries to it. */
if (c)
repl_remove_standby(&c->persist.repl, ep);
free(sctx);
ep->data = NULL;
return 0;

50
src/controller/repl.c
View file

@ -12,6 +12,15 @@
#include "repl.h"
#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
@ -69,6 +78,15 @@ int repl_add_standby(replicator_t *r, repl_ep_t ep, const char *host, uint32_t p
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;
@ -187,15 +205,11 @@ void repl_on_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_lsn)
* 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).
* TCP send wiring deferred to task 7.6.
* ============================================================ */
int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
const void *payload, uint32_t payload_len)
{
(void)payload;
(void)payload_len;
/* WAL_CHECKPOINT entries must never be replicated */
assert(hdr != NULL);
assert(hdr->type != WAL_CHECKPOINT);
@ -225,7 +239,33 @@ int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
continue;
}
/* Non-evicted standby: track the pending send (TCP wiring in task 7.6) */
/* Build a single-entry batch and enqueue it for this standby. */
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;
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++;
}

232
src/controller/testsuite/test_repl_unit.c
View file

@ -9,6 +9,18 @@
#include <unity/unity.h>
#include "repl.h"
#include "config.h"
#include "typemap.h"
#include "wal_types.h"
/* ============================================================
* Test seam for repl_broadcast send path (task 1.1)
* Declared as extern here; defined in repl.c under XCE_TESTING.
* ============================================================ */
#ifdef XCE_TESTING
extern int (*repl_broadcast_send_hook)(void *ep,
wal_catchup_entry_t **entries,
int n_entries);
#endif
static void test_repl_init_zeroes_nstandbys(void) {
replicator_t r;
@ -160,6 +172,218 @@ static void test_wait_ack_woken_by_background_ack(void) {
printf("test_wait_ack_woken_by_background_ack: PASS\n");
}
/* ============================================================
* Task 1.1: repl_broadcast send path tests
* ============================================================ */
/* Shared hook state for send-path tests */
static int g_hook_call_count;
static void *g_hook_last_ep;
static uint64_t g_hook_last_lsn;
static uint8_t g_hook_last_entry_type;
static const void *g_hook_last_payload;
static size_t g_hook_last_payload_len;
static int g_hook_return_val; /* 0 = success, negative = error */
static int test_broadcast_hook(void *ep, wal_catchup_entry_t **entries,
int n_entries)
{
g_hook_call_count++;
g_hook_last_ep = ep;
if (n_entries >= 1 && entries && entries[0]) {
g_hook_last_lsn = entries[0]->lsn;
g_hook_last_entry_type = entries[0]->entry_type;
g_hook_last_payload = entries[0]->raw_payload;
g_hook_last_payload_len = entries[0]->raw_payload_len;
}
return g_hook_return_val;
}
static void reset_hook_state(void)
{
g_hook_call_count = 0;
g_hook_last_ep = NULL;
g_hook_last_lsn = 0;
g_hook_last_entry_type = 0;
g_hook_last_payload = NULL;
g_hook_last_payload_len = 0;
g_hook_return_val = 0;
}
/* Test: no standbys — broadcast returns 0, hook not called */
static void test_broadcast_no_standbys_returns_zero(void)
{
#ifdef XCE_TESTING
replicator_t r;
repl_init(&r);
reset_hook_state();
repl_broadcast_send_hook = test_broadcast_hook;
wal_entry_hdr_t hdr = { .lsn = 10, .type = WAL_JOB_SUBMIT };
uint8_t payload[] = { 0xAA, 0xBB };
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload));
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(0, g_hook_call_count);
repl_broadcast_send_hook = NULL;
repl_fini(&r);
printf("test_broadcast_no_standbys_returns_zero: PASS\n");
#else
printf("test_broadcast_no_standbys_returns_zero: SKIPPED (no XCE_TESTING)\n");
#endif
}
/* Test: one standby — hook called once with correct lsn, type, payload */
static void test_broadcast_one_standby_sends_entry(void)
{
#ifdef XCE_TESTING
replicator_t r;
int dummy_ep = 0;
repl_init(&r);
repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001);
reset_hook_state();
repl_broadcast_send_hook = test_broadcast_hook;
wal_entry_hdr_t hdr = { .lsn = 42, .type = WAL_JOB_SUBMIT };
uint8_t payload[] = { 1, 2, 3, 4 };
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload));
TEST_ASSERT_EQUAL_INT(0, rc);
TEST_ASSERT_EQUAL_INT(1, g_hook_call_count);
TEST_ASSERT_EQUAL_PTR((void *)&dummy_ep, g_hook_last_ep);
TEST_ASSERT_EQUAL_UINT64(42, g_hook_last_lsn);
TEST_ASSERT_EQUAL_UINT8(WAL_JOB_SUBMIT, g_hook_last_entry_type);
TEST_ASSERT_EQUAL_PTR(payload, g_hook_last_payload);
TEST_ASSERT_EQUAL_UINT(sizeof(payload), g_hook_last_payload_len);
repl_broadcast_send_hook = NULL;
repl_fini(&r);
printf("test_broadcast_one_standby_sends_entry: PASS\n");
#else
printf("test_broadcast_one_standby_sends_entry: SKIPPED (no XCE_TESTING)\n");
#endif
}
/* Test: standby in catch-up is skipped — hook not called for it */
static void test_broadcast_skips_catchup_standby(void)
{
#ifdef XCE_TESTING
replicator_t r;
int ep1 = 0, ep2 = 1;
repl_init(&r);
repl_add_standby(&r, (void *)&ep1, "sb1", 7001);
repl_add_standby(&r, (void *)&ep2, "sb2", 7002);
/* Mark sb1 as in catch-up */
repl_start_catchup(&r, (void *)&ep1, 0);
reset_hook_state();
repl_broadcast_send_hook = test_broadcast_hook;
wal_entry_hdr_t hdr = { .lsn = 99, .type = WAL_JOB_SUBMIT };
uint8_t payload[] = { 0xFF };
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload));
TEST_ASSERT_EQUAL_INT(0, rc);
/* Only sb2 should be sent to */
TEST_ASSERT_EQUAL_INT(1, g_hook_call_count);
TEST_ASSERT_EQUAL_PTR((void *)&ep2, g_hook_last_ep);
repl_broadcast_send_hook = NULL;
repl_fini(&r);
printf("test_broadcast_skips_catchup_standby: PASS\n");
#else
printf("test_broadcast_skips_catchup_standby: SKIPPED (no XCE_TESTING)\n");
#endif
}
/* Test: standby with pending_ack_count > MAX_PENDING_ACKS is evicted to catch-up */
static void test_broadcast_evicts_overloaded_standby(void)
{
#ifdef XCE_TESTING
replicator_t r;
int dummy_ep = 0;
repl_init(&r);
int idx = repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001);
/* Manually set pending_ack_count just above the limit */
r.standbys[idx].pending_ack_count = MAX_PENDING_ACKS + 1;
reset_hook_state();
repl_broadcast_send_hook = test_broadcast_hook;
wal_entry_hdr_t hdr = { .lsn = 5, .type = WAL_JOB_SUBMIT };
uint8_t payload[] = { 0x01 };
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload));
TEST_ASSERT_EQUAL_INT(0, rc);
/* Evicted — hook must NOT be called */
TEST_ASSERT_EQUAL_INT(0, g_hook_call_count);
/* Standby must now be in catch-up */
TEST_ASSERT_NOT_EQUAL(0,
atomic_load_explicit(&r.standbys[idx].in_catch_up, memory_order_acquire));
repl_broadcast_send_hook = NULL;
repl_fini(&r);
printf("test_broadcast_evicts_overloaded_standby: PASS\n");
#else
printf("test_broadcast_evicts_overloaded_standby: SKIPPED (no XCE_TESTING)\n");
#endif
}
/* Test (Task 2.3): repl_add_standby is idempotent for the same ep.
* Calling it twice with the same endpoint must return the same index,
* keep nstandbys==1, and not crash or corrupt state.
* Req 1.3: If repl_add_standby is called with an endpoint already
* registered, the system shall treat it as idempotent (no-op or update-in-place). */
static void test_repl_add_standby_idempotent(void) {
replicator_t r;
int dummy_ep = 0;
repl_init(&r);
int idx1 = repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001);
TEST_ASSERT_TRUE(idx1 >= 0);
TEST_ASSERT_EQUAL_INT(1, r.nstandbys);
/* Second call with the same ep — must be a no-op */
int idx2 = repl_add_standby(&r, (void *)&dummy_ep, "sb1", 7001);
TEST_ASSERT_TRUE(idx2 >= 0);
/* nstandbys must still be 1 (not 2) */
TEST_ASSERT_EQUAL_INT(1, r.nstandbys);
/* Must return the same slot index */
TEST_ASSERT_EQUAL_INT(idx1, idx2);
repl_fini(&r);
printf("test_repl_add_standby_idempotent: PASS\n");
}
/* Test: send failure (hook returns -EIO) is tolerated — broadcast returns 0 */
static void test_broadcast_send_failure_continues(void)
{
#ifdef XCE_TESTING
replicator_t r;
int ep1 = 0, ep2 = 1;
repl_init(&r);
repl_add_standby(&r, (void *)&ep1, "sb1", 7001);
repl_add_standby(&r, (void *)&ep2, "sb2", 7002);
reset_hook_state();
g_hook_return_val = -EIO; /* First call fails */
repl_broadcast_send_hook = test_broadcast_hook;
wal_entry_hdr_t hdr = { .lsn = 77, .type = WAL_JOB_SUBMIT };
uint8_t payload[] = { 0xDE, 0xAD };
int rc = repl_broadcast(&r, &hdr, payload, sizeof(payload));
/* broadcast should still return 0 — errors are WARN+continue */
TEST_ASSERT_EQUAL_INT(0, rc);
/* Both standbys attempted */
TEST_ASSERT_EQUAL_INT(2, g_hook_call_count);
repl_broadcast_send_hook = NULL;
repl_fini(&r);
printf("test_broadcast_send_failure_continues: PASS\n");
#else
printf("test_broadcast_send_failure_continues: SKIPPED (no XCE_TESTING)\n");
#endif
}
void test_repl_unit(void) {
UNITY_BEGIN();
RUN_TEST(test_repl_init_zeroes_nstandbys);
@ -171,5 +395,13 @@ void test_repl_unit(void) {
RUN_TEST(test_wait_ack_already_acked);
RUN_TEST(test_wait_ack_timeout);
RUN_TEST(test_wait_ack_woken_by_background_ack);
/* Task 2.3: idempotent re-registration */
RUN_TEST(test_repl_add_standby_idempotent);
/* Task 1.1: broadcast send path */
RUN_TEST(test_broadcast_no_standbys_returns_zero);
RUN_TEST(test_broadcast_one_standby_sends_entry);
RUN_TEST(test_broadcast_skips_catchup_standby);
RUN_TEST(test_broadcast_evicts_overloaded_standby);
RUN_TEST(test_broadcast_send_failure_continues);
UNITY_END();
}