From 81dbc4d7b4fe4eaf99dbedccaa7e61249f5f5a0b Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 10 Jun 2026 08:35:02 +0000 Subject: [PATCH] =?UTF-8?q?feat(ha-wal-replication):=20HA=E2=86=92replicat?= =?UTF-8?q?or=20standby=20registration=20bridge=20(tasks=202.1-2.3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .kiro/specs/ha-wal-replication/tasks.md | 124 ++++++++++++ src/controller/ha.c | 154 ++++++++++++-- src/controller/repl.c | 50 ++++- src/controller/testsuite/test_repl_unit.c | 232 ++++++++++++++++++++++ 4 files changed, 542 insertions(+), 18 deletions(-) create mode 100644 .kiro/specs/ha-wal-replication/tasks.md diff --git a/.kiro/specs/ha-wal-replication/tasks.md b/.kiro/specs/ha-wal-replication/tasks.md new file mode 100644 index 00000000..4ddc11e9 --- /dev/null +++ b/.kiro/specs/ha-wal-replication/tasks.md @@ -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)_ diff --git a/src/controller/ha.c b/src/controller/ha.c index 369c559e..55044e23 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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; diff --git a/src/controller/repl.c b/src/controller/repl.c index 41079e83..e0820fb0 100644 --- a/src/controller/repl.c +++ b/src/controller/repl.c @@ -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++; } diff --git a/src/controller/testsuite/test_repl_unit.c b/src/controller/testsuite/test_repl_unit.c index 98de0404..36863f61 100644 --- a/src/controller/testsuite/test_repl_unit.c +++ b/src/controller/testsuite/test_repl_unit.c @@ -9,6 +9,18 @@ #include #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(); }