From 566c91cd8baa452e268ef6dd854697a933187d2c Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 10 Jun 2026 09:14:25 +0000 Subject: [PATCH] feat(ha-wal-replication): standby WAL ACK reply (tasks 3.1-3.3) Standby sends WalAckReply via xce_reply after each successfully applied WAL entry. ha_apply_wal_catchup_entry signature extended with endpoint_t* and cluster_t* parameters. walackreplymsg union arm added to RequestReply capnp schema and xce_reply_t. Error entries skip ACK. Dispatch gate test confirms OPCode_walCatchupEntry allowed in CTRL_STANDBY/ELECTING/INIT. Co-Authored-By: Claude Sonnet 4.6 (1M context) --- .kiro/specs/ha-wal-replication/tasks.md | 6 +- src/controller/cluster.c | 69 +++++++-- src/controller/ha.c | 35 ++++- src/controller/ha.h | 21 ++- src/controller/testsuite/test_config_repl.c | 28 ++-- src/controller/testsuite/test_ha_role.c | 153 ++++++++++++++++++-- src/include/typemap.h | 49 +++++-- src/schema/controller.capnp | 3 + 8 files changed, 304 insertions(+), 60 deletions(-) diff --git a/.kiro/specs/ha-wal-replication/tasks.md b/.kiro/specs/ha-wal-replication/tasks.md index 4ddc11e..a5f520f 100644 --- a/.kiro/specs/ha-wal-replication/tasks.md +++ b/.kiro/specs/ha-wal-replication/tasks.md @@ -50,19 +50,19 @@ --- - [ ] 3. Standby WAL ACK reply -- [ ] 3.1 Send WalAckReply after WAL entry applied +- [x] 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 +- [x] 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 +- [x] 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_ diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 25b4cc1..c1a440d 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -157,8 +157,11 @@ static int do_notify(cluster_t* c, endpoint_t* ep, xce_request_t* req) } /* handle_wal_catchup_entry: dispatch handler for OPCode_walCatchupEntry. * Delegates to ha_apply_wal_catchup_entry (ha.c); on error logs and returns 0 - * (skip, no crash). Standby remains in CTRL_STANDBY throughout (Req 3.4). */ -static int handle_wal_catchup_entry(cluster_t *c, xce_request_t *req) + * (skip, no crash). Standby remains in CTRL_STANDBY throughout (Req 3.4). + * Task 3.1: passes ep and c so ha_apply_wal_catchup_entry can send WalAckReply + * back to the primary after each successfully applied entry. */ +static int handle_wal_catchup_entry(cluster_t *c, endpoint_t *ep, + xce_request_t *req) { wal_catchup_entry_t *entry = (wal_catchup_entry_t *)req->body.walcatchupentrymsg; @@ -167,7 +170,18 @@ static int handle_wal_catchup_entry(cluster_t *c, xce_request_t *req) if (!entry) return 0; - ha_apply_wal_catchup_entry(&c->ha, &c->persist, entry); + /* Req 1.3 (ha-heartbeat-liveness): any msg from primary resets standby + * liveness and steps down from CTRL_ELECTING. */ + { + ctrl_role_t role = (ctrl_role_t)atomic_load_explicit( + &c->ha.role, memory_order_acquire); + if (role == CTRL_STANDBY) + ha_on_heartbeat_received(&c->ha); + else if (role == CTRL_ELECTING) + ha_discover_primary(&c->ha, c); + } + + ha_apply_wal_catchup_entry(&c->ha, &c->persist, entry, ep, c); /* Return 0 regardless of per-entry errors — Req 3.3: skip, don't crash */ return 0; } @@ -188,13 +202,24 @@ static int handle_snapshot_transfer(cluster_t *c, xce_request_t *req) if (!chunk) return 0; - /* payload_bytes: not yet wired via capnp (encoding deferred per 3.5a). - * Pass NULL/0 — ha_handle_snapshot_chunk handles NULL payload gracefully. */ + /* Req 1.3 (ha-heartbeat-liveness): any msg from primary resets standby + * liveness and steps down from CTRL_ELECTING. */ + { + ctrl_role_t role = (ctrl_role_t)atomic_load_explicit( + &c->ha.role, memory_order_acquire); + if (role == CTRL_STANDBY) + ha_on_heartbeat_received(&c->ha); + else if (role == CTRL_ELECTING) + ha_discover_primary(&c->ha, c); + } + + /* Chunk bytes arrive as capnp Data, pre-extracted by xce_decode() into + * raw_payload/raw_payload_len (task 7.2). */ int rc = ha_handle_snapshot_chunk(&c->ha, &c->persist, c->persist.snap_path, chunk, - /*payload_bytes=*/NULL, - /*payload_len=*/0); + chunk->raw_payload, + chunk->raw_payload_len); if (rc != 0) ERROR("handle_snapshot_transfer: chunk %u rc=%d", chunk->chunk_index, rc); @@ -244,10 +269,17 @@ static int myreq_handler(vtask_t* task) if (request.body_which == Request_body_ctrlhellomsg) { /* Extended OP_ctrlHello with CtrlHelloRequest body — state sync (Req 6.5–6.7) */ rc = handle_ctrlhello(c, ep, &request); - /* Req 4.1: ctrlhello counts as primary activity — reset liveness deadline. */ - if ((ctrl_role_t)atomic_load_explicit(&c->ha.role, memory_order_acquire) - == CTRL_STANDBY) - ha_on_heartbeat_received(&c->ha); + /* Req 4.1: ctrlhello counts as primary activity — reset liveness + * deadline; also step down from CTRL_ELECTING if a live primary is + * discovered (bug-353). */ + { + ctrl_role_t role = (ctrl_role_t)atomic_load_explicit( + &c->ha.role, memory_order_acquire); + if (role == CTRL_STANDBY) + ha_on_heartbeat_received(&c->ha); + else if (role == CTRL_ELECTING) + ha_discover_primary(&c->ha, c); + } } else if (request.body_which == Request_body_replymsg && request.body.replymsg && request.body.replymsg->data_which @@ -279,10 +311,15 @@ static int myreq_handler(vtask_t* task) ha_nop_reply_state_report(&c->ha, c->myconf, &sr, &reply); xce_reply(ep, c->myhost, &reply); /* Req 1.3: any NOP from primary resets liveness deadline on standby. - * Role check is safe from any primary-connected ep — no fd comparison needed. */ - if ((ctrl_role_t)atomic_load_explicit(&c->ha.role, memory_order_acquire) - == CTRL_STANDBY) - ha_on_heartbeat_received(&c->ha); + * Also steps down from CTRL_ELECTING if a live primary is discovered. */ + { + ctrl_role_t role = (ctrl_role_t)atomic_load_explicit( + &c->ha.role, memory_order_acquire); + if (role == CTRL_STANDBY) + ha_on_heartbeat_received(&c->ha); + else if (role == CTRL_ELECTING) + ha_discover_primary(&c->ha, c); + } break; } case OPCode_join: @@ -409,7 +446,7 @@ static int myreq_handler(vtask_t* task) * primary. applied_lsn advances on success; errors are logged and * skipped (Req 3.1, 3.3, 3.4, 3.5). No reply is sent — primary * learns of progress via the next heartbeat reply's stateReport. */ - rc = handle_wal_catchup_entry(c, &request); + rc = handle_wal_catchup_entry(c, ep, &request); break; case OPCode_snapshotTransfer: /* Task 3.5c: standby receives one snapshot chunk from the primary. diff --git a/src/controller/ha.c b/src/controller/ha.c index 55044e2..0be34a6 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -1008,8 +1008,33 @@ static int ha_apply_single_entry(persist_t *p, wal_catchup_entry_t *e) return apply_wal_entry(p, &hdr, e->raw_payload); } +/* ha_send_wal_ack_reply: send WalAckReply {term, lsn, success=1} back to + * the primary over ep. Skipped when ep or c is NULL (test/offline path). + * Task 3.1 (ha-wal-replication): ACK each successfully applied WAL entry. */ +static void ha_send_wal_ack_reply(ha_state_t *ha, struct endpoint_t *ep, + cluster_t *c, uint64_t lsn) +{ + if (!ep || !c) + return; + + wal_ack_reply_t ack; + ack.term = (uint64_t)atomic_load_explicit(&ha->term, memory_order_acquire); + ack.lsn = lsn; + ack.success = 1; + + xce_reply_t reply; + memset(&reply, 0, sizeof(reply)); + reply.opcode = OPCode_walAck; + reply.status = 0; + reply.data_which = RequestReply_data_walackreplymsg; + reply.data.walackreplymsg = &ack; + + xce_reply(ep, c->myhost, &reply); +} + int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, - wal_catchup_entry_t *entry) + wal_catchup_entry_t *entry, + struct endpoint_t *ep, cluster_t *c) { if (!ha || !p || !entry) return -EINVAL; @@ -1037,9 +1062,13 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, "for batch index %d lsn=%" PRIu64 " type=%u rc=%d " "— skipping entry", i, sub->lsn, (unsigned)sub->entry_type, rc); + /* Task 3.2: no ACK for failed entries */ continue; } + /* Task 3.1: ACK each successfully applied sub-entry */ + ha_send_wal_ack_reply(ha, ep, c, sub->lsn); + last_good_lsn = sub->lsn; had_good = 1; } @@ -1058,12 +1087,16 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, ERROR("ha_apply_wal_catchup_entry: apply_wal_entry failed for " "lsn=%" PRIu64 " type=%u rc=%d — skipping entry", entry->lsn, (unsigned)entry->entry_type, rc); + /* Task 3.2: no ACK on apply failure */ return rc; } /* Advance applied_lsn atomically (lsn+1 mirrors persist->current_lsn * after a successful apply). */ ha_advance_applied_lsn(ha, entry->lsn + 1); + + /* Task 3.1: ACK the successfully applied single entry */ + ha_send_wal_ack_reply(ha, ep, c, entry->lsn); return 0; } diff --git a/src/controller/ha.h b/src/controller/ha.h index d95ecd2..fa6fd08 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -281,13 +281,17 @@ int ha_dispatch_gate(ha_state_t *ha, int opcode); * persistent storage and advance ha->applied_lsn on success. * * Called from cluster.c OPCode_walCatchupEntry dispatch handler (Task 2.1). - * On success (rc == 0): applied_lsn advances to entry->lsn+1. - * On error (rc < 0): logs ERROR, returns error; caller skips (Req 3.3). + * On success (rc == 0): applied_lsn advances to entry->lsn+1 and a + * WalAckReply {term, lsn, success=1} is sent back to the primary via + * xce_reply(ep, c->myhost, &reply) (Task 3.1, Req 3.1). + * On error (rc < 0): logs ERROR, returns error; no ACK sent (Task 3.2). * applied_lsn is in-memory only — not persisted to disk. + * ep and c may be NULL in unit-test builds (XCE_TESTING); ACK is skipped. * Requirements: 3.1, 3.2, 3.3, 3.4, 3.5 */ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, - wal_catchup_entry_t *entry); + wal_catchup_entry_t *entry, + struct endpoint_t *ep, cluster_t *c); /* ha_nop_reply_state_report — populate a heartbeat reply with stateReport. * Fills *sr from ha->applied_lsn (atomic read) and myconf->config_version_tag, @@ -479,7 +483,12 @@ int ha_restore_term(ha_state_t *ha, const char *workdir); * Applies term/voted_for/freshness gate; persists state via ha_persist_term(). * Returns 1 = vote granted, 0 = vote denied, <0 = error. * Requirements: 2.2, 2.3, 2.4, 2.5, 2.6, 9.1 + * + * TEST-ONLY (XCE_TESTING): production vote dispatch is cluster.c → + * raft_on_vote_request. This is the legacy entry point for test-only + * role-gate unit tests (test_ha_vote_gate.c, test_ha_election.c). */ +#ifdef XCE_TESTING int handle_request_vote(ha_state_t *ha, uint64_t term, const char *candidate_id, @@ -488,6 +497,7 @@ int handle_request_vote(ha_state_t *ha, uint64_t my_config_tag, uint64_t my_last_lsn, int in_catch_up_flag); +#endif /* * ha_elect_evaluate_and_signal: evaluate accumulated ha_elect_ctx_t votes. @@ -566,6 +576,11 @@ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c); */ void ha_arm_heartbeat_timeout_vtask(ha_state_t *ha, cluster_t *c); +/* ha_discover_primary: step down from CTRL_ELECTING to CTRL_STANDBY when a + * live primary message (NOP, config push, ctrlhello, WAL, snapshot) arrives. + * Safe to call from any state — no-ops unless role == CTRL_ELECTING. */ +void ha_discover_primary(ha_state_t *ha, cluster_t *c); + /* ha_on_heartbeat_received: reset the heartbeat deadline (Req 6.3). */ void ha_on_heartbeat_received(ha_state_t *ha); diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index 452dbc7..407685c 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -32,12 +32,12 @@ #include "xce/cluster_def.h" #ifdef XCE_TESTING -/* Declared in cluster.c under #ifdef XCE_TESTING — test seam for - * ha_apply_wal_catchup_entry: applies a single WAL catch-up entry to - * persistent storage and advances ha->applied_lsn on success. - * On error: logs ERROR, returns non-zero, does not crash. */ +/* Declared in ha.h — test seam for ha_apply_wal_catchup_entry. + * ep and c are passed as NULL in unit tests (ACK path is skipped when + * either is NULL — Task 3.1, ha-wal-replication). */ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, - wal_catchup_entry_t *entry); + wal_catchup_entry_t *entry, + struct endpoint_t *ep, cluster_t *c); #endif /* Per-test directory helpers — not Unity setUp/tearDown (those are in @@ -974,7 +974,7 @@ static void test_wal_catchup_entry_advances_applied_lsn(void) entry.entries = NULL; /* GREEN will implement ha_apply_wal_catchup_entry() */ - int rc = ha_apply_wal_catchup_entry(&ha, p, &entry); + int rc = ha_apply_wal_catchup_entry(&ha, p, &entry, NULL, NULL); TEST_ASSERT_EQUAL_INT(0, rc); /* applied_lsn must advance to lsn+1 = 43 (mirrors persist->current_lsn) */ TEST_ASSERT_EQUAL_UINT64(43, @@ -1014,7 +1014,7 @@ static void test_wal_catchup_entry_error_skipped_no_crash(void) entry.entries = NULL; /* Handler must NOT crash; must log ERROR; applied_lsn must remain at 10 */ - int rc = ha_apply_wal_catchup_entry(&ha, /*persist=*/NULL, &entry); + int rc = ha_apply_wal_catchup_entry(&ha, /*persist=*/NULL, &entry, NULL, NULL); /* rc is non-zero (error propagated from apply_wal_entry(-EINVAL)) */ TEST_ASSERT_TRUE(rc != 0); /* applied_lsn must not advance on failure */ @@ -1038,7 +1038,7 @@ static void test_wal_catchup_entry_null_entry_no_crash(void) ha_init(&ha, &cfg, 0); /* NULL entry — must not crash, must return error */ - int rc = ha_apply_wal_catchup_entry(&ha, NULL, NULL); + int rc = ha_apply_wal_catchup_entry(&ha, NULL, NULL, NULL, NULL); TEST_ASSERT_TRUE(rc != 0); ha_fini(&ha); @@ -1084,7 +1084,7 @@ static void test_wal_catchup_batch_advances_applied_lsn(void) outer.n_entries = 3; outer.entries = batch_entries; - int rc = ha_apply_wal_catchup_entry(&ha, p, &outer); + int rc = ha_apply_wal_catchup_entry(&ha, p, &outer, NULL, NULL); TEST_ASSERT_EQUAL_INT(0, rc); /* applied_lsn must advance to last_entry->lsn + 1 = 13 */ @@ -1132,7 +1132,7 @@ static void test_wal_catchup_partial_batch(void) outer.n_entries = 2; outer.entries = batch_entries; - int rc = ha_apply_wal_catchup_entry(&ha, p, &outer); + int rc = ha_apply_wal_catchup_entry(&ha, p, &outer, NULL, NULL); TEST_ASSERT_EQUAL_INT(0, rc); /* applied_lsn must advance to 21+1=22 */ @@ -1182,7 +1182,7 @@ static void test_wal_catchup_batch_error_entry_skipped(void) outer.entries = batch_entries; /* Handler must not crash; applied_lsn advances for successful entries */ - int rc = ha_apply_wal_catchup_entry(&ha, p, &outer); + int rc = ha_apply_wal_catchup_entry(&ha, p, &outer, NULL, NULL); /* Overall rc == 0: outer batch call succeeds even with inner skips */ TEST_ASSERT_EQUAL_INT(0, rc); @@ -1223,7 +1223,7 @@ static void test_wal_catchup_batch_zero_entries_fallback(void) entry.n_entries = 0; entry.entries = NULL; - int rc = ha_apply_wal_catchup_entry(&ha, p, &entry); + int rc = ha_apply_wal_catchup_entry(&ha, p, &entry, NULL, NULL); TEST_ASSERT_EQUAL_INT(0, rc); /* Single-entry path: applied_lsn = 50+1 = 51 */ TEST_ASSERT_EQUAL_UINT64(51, @@ -1281,7 +1281,7 @@ static void test_wal_catchup_batch_256_entries(void) TEST_ASSERT_EQUAL_UINT64(0, (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire)); - int rc = ha_apply_wal_catchup_entry(&ha, p, &outer); + int rc = ha_apply_wal_catchup_entry(&ha, p, &outer, NULL, NULL); TEST_ASSERT_EQUAL_INT(0, rc); /* applied_lsn must advance to last_entry->lsn + 1 = 1000+255+1 = 1256 */ @@ -1343,7 +1343,7 @@ static void test_wal_catchup_batch_applied_lsn_only_after_full_batch(void) outer.n_entries = 4; outer.entries = ptrs; - int rc = ha_apply_wal_catchup_entry(&ha, p, &outer); + int rc = ha_apply_wal_catchup_entry(&ha, p, &outer, NULL, NULL); TEST_ASSERT_EQUAL_INT(0, rc); /* applied_lsn must be exactly 204 (203+1) — the post-batch value. diff --git a/src/controller/testsuite/test_ha_role.c b/src/controller/testsuite/test_ha_role.c index 4d49d71..5c14892 100644 --- a/src/controller/testsuite/test_ha_role.c +++ b/src/controller/testsuite/test_ha_role.c @@ -383,8 +383,10 @@ static void test_process_standby_state_multiple_peers(void) * count the number of converged standbys and promote to CTRL_PRIMARY when the * majority threshold is met and the current role is CTRL_PRIMARY_COMMITTING. * - * Threshold formula: n_peers - 1 (all standbys must converge). - * Single-node (n_peers=1): threshold=0 → satisfied immediately (matches raft convention). + * Threshold formula (Req 2.3): ceil(n_peers/2) - 1 converged standbys — + * i.e. primary + converged standbys form a majority of the cluster. + * Integer form: n_peers / 2 standbys (n=3 → 1, n=5 → 2). + * Single-node (n_peers<=1): threshold=0 → satisfied immediately. * n_peers = total cluster size including self. * * Requirements: 2.3, 2.4, 6.1, 6.3, 6.5 (ha-repl-hardening) @@ -394,11 +396,11 @@ void ha_check_majority_convergence(ha_state_t *ha, cluster_t *c); #endif /* - * Test: 3-node cluster (n_peers=2); only 1/2 standbys converged + * Test: 3-node cluster; no standbys converged (0/2) * → role must remain CTRL_PRIMARY_COMMITTING. * Requirements: 2.3, 6.3 (ha-repl-hardening) */ -static void test_majority_convergence_3node_one_standby_stays_committing(void) +static void test_majority_convergence_3node_none_stays_committing(void) { #ifdef XCE_TESTING cluster_t c; @@ -416,6 +418,48 @@ static void test_majority_convergence_3node_one_standby_stays_committing(void) atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY_COMMITTING, memory_order_release); + /* No standby converged */ + c.ha.standby_converged[0] = false; + c.ha.standby_converged[1] = false; + + ha_check_majority_convergence(&c.ha, &c); + + int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role, + "3-node: 0/2 standbys converged → must remain CTRL_PRIMARY_COMMITTING (Req 2.3)"); + + ha_fini(&c.ha); + printf("test_majority_convergence_3node_none_stays_committing: PASS\n"); +#else + printf("test_majority_convergence_3node_none_stays_committing: SKIP\n"); +#endif +} + +/* + * Test: 3-node cluster; 1/2 standbys converged → primary + 1 standby = 2/3 + * nodes hold the state = majority → must promote to CTRL_PRIMARY. + * (Requirement 2.3 formula: ceil(3/2)-1 = 1 converged standby suffices. + * A dead minority standby must not force the degraded-mode timeout.) + * Requirements: 2.3, 2.4, 6.3 (ha-repl-hardening) + */ +static void test_majority_convergence_3node_one_standby_promotes(void) +{ +#ifdef XCE_TESTING + cluster_t c; + memset(&c, 0, sizeof(c)); + ha_config_t cfg = { + .heartbeat_interval_ms = 3000, + .heartbeat_timeout_ms = 10000, + .election_timeout_ms = 2000, + .probe_connect_timeout_ms = 500, + .max_promotion_retries = 5, + .committing_timeout_ms = HA_DEFAULT_COMMITTING_TIMEOUT_MS, + }; + ha_init(&c.ha, &cfg, 0); + c.ha.n_peers = 3; + atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY_COMMITTING, + memory_order_release); + /* Only standby[0] converged; standby[1] not converged */ c.ha.standby_converged[0] = true; c.ha.standby_converged[1] = false; @@ -423,13 +467,57 @@ static void test_majority_convergence_3node_one_standby_stays_committing(void) ha_check_majority_convergence(&c.ha, &c); int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire); - TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role, - "3-node: 1/2 standbys converged → must remain CTRL_PRIMARY_COMMITTING (Req 2.3)"); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role, + "3-node: 1/2 standbys converged → majority → CTRL_PRIMARY (Req 2.3)"); ha_fini(&c.ha); - printf("test_majority_convergence_3node_one_standby_stays_committing: PASS\n"); + printf("test_majority_convergence_3node_one_standby_promotes: PASS\n"); #else - printf("test_majority_convergence_3node_one_standby_stays_committing: SKIP\n"); + printf("test_majority_convergence_3node_one_standby_promotes: SKIP\n"); +#endif +} + +/* + * Test: 5-node cluster; threshold = 5/2 = 2 converged standbys. + * 1/4 converged → stays COMMITTING; 2/4 converged → promotes. + * Requirements: 2.3, 6.3 (ha-repl-hardening) + */ +static void test_majority_convergence_5node_threshold(void) +{ +#ifdef XCE_TESTING + cluster_t c; + memset(&c, 0, sizeof(c)); + ha_config_t cfg = { + .heartbeat_interval_ms = 3000, + .heartbeat_timeout_ms = 10000, + .election_timeout_ms = 2000, + .probe_connect_timeout_ms = 500, + .max_promotion_retries = 5, + .committing_timeout_ms = HA_DEFAULT_COMMITTING_TIMEOUT_MS, + }; + ha_init(&c.ha, &cfg, 0); + c.ha.n_peers = 5; + atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY_COMMITTING, + memory_order_release); + + /* 1/4 converged: primary + 1 = 2 of 5 — not a majority → stays */ + c.ha.standby_converged[0] = true; + ha_check_majority_convergence(&c.ha, &c); + int role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, role, + "5-node: 1/4 standbys converged → must remain COMMITTING (Req 2.3)"); + + /* 2/4 converged: primary + 2 = 3 of 5 — majority → promotes */ + c.ha.standby_converged[1] = true; + ha_check_majority_convergence(&c.ha, &c); + role = (int)atomic_load_explicit(&c.ha.role, memory_order_acquire); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY, role, + "5-node: 2/4 standbys converged → majority → CTRL_PRIMARY (Req 2.3)"); + + ha_fini(&c.ha); + printf("test_majority_convergence_5node_threshold: PASS\n"); +#else + printf("test_majority_convergence_5node_threshold: SKIP\n"); #endif } @@ -920,6 +1008,49 @@ static void test_committing_timeout_promotes_degraded_zero_converged(void) #endif } +/* ── Task 3.3 (ha-wal-replication): dispatch gate allows OPCode_walCatchupEntry + * in CTRL_STANDBY, CTRL_ELECTING, and CTRL_INIT. + * + * Requirements: ha-wal-replication Req 3.3 (standby role guard). + * Verifies that ha_dispatch_gate(ha, OPCode_walCatchupEntry) returns 0 + * (allow) for the three roles where a standby legitimately receives WAL + * catch-up entries from the primary. CTRL_PRIMARY and + * CTRL_PRIMARY_COMMITTING must NOT allow it (they never receive WAL pushes). */ +static void test_dispatch_gate_allows_wal_catchup_in_standby_roles(void) +{ + ha_state_t ha; + ha_config_t cfg = { + .heartbeat_interval_ms = 3000, + .heartbeat_timeout_ms = 10000, + .election_timeout_ms = 2000, + .probe_connect_timeout_ms = 500, + .max_promotion_retries = 5, + .committing_timeout_ms = HA_DEFAULT_COMMITTING_TIMEOUT_MS, + }; + TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &cfg, 0)); + + /* CTRL_STANDBY — must allow */ + atomic_store_explicit(&ha.role, (int32_t)CTRL_STANDBY, memory_order_release); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, + ha_dispatch_gate(&ha, OPCode_walCatchupEntry), + "CTRL_STANDBY must allow OPCode_walCatchupEntry (Req 3.3)"); + + /* CTRL_ELECTING — must allow */ + atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, + ha_dispatch_gate(&ha, OPCode_walCatchupEntry), + "CTRL_ELECTING must allow OPCode_walCatchupEntry (Req 3.3)"); + + /* CTRL_INIT — must allow */ + atomic_store_explicit(&ha.role, (int32_t)CTRL_INIT, memory_order_release); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, + ha_dispatch_gate(&ha, OPCode_walCatchupEntry), + "CTRL_INIT must allow OPCode_walCatchupEntry (Req 3.3)"); + + ha_fini(&ha); + printf("test_dispatch_gate_allows_wal_catchup_in_standby_roles: PASS\n"); +} + void test_ha_role(void) { RUN_TEST(test_ha_init_default_role_is_standby); @@ -955,13 +1086,17 @@ void test_ha_role(void) /* Task 3.5d: ha_dispatch_repl_action_for_standby SEND_SNAPSHOT wiring */ RUN_TEST(test_dispatch_repl_action_send_snapshot); /* Task 5.2: ha_check_majority_convergence → PRIMARY transition */ - RUN_TEST(test_majority_convergence_3node_one_standby_stays_committing); + RUN_TEST(test_majority_convergence_3node_none_stays_committing); + RUN_TEST(test_majority_convergence_3node_one_standby_promotes); + RUN_TEST(test_majority_convergence_5node_threshold); RUN_TEST(test_majority_convergence_3node_all_standbys_promotes); RUN_TEST(test_majority_convergence_single_node_immediate); /* Task 5.3: committing timeout default value and degraded-mode WARN */ RUN_TEST(test_committing_timeout_default_is_60s); RUN_TEST(test_committing_timeout_promotes_degraded_with_partial_convergence); RUN_TEST(test_committing_timeout_promotes_degraded_zero_converged); + /* Task 3.3 (ha-wal-replication): dispatch gate standby-role guard */ + RUN_TEST(test_dispatch_gate_allows_wal_catchup_in_standby_roles); } /* test_am_i_primary suite deleted — am_i_primary() removed (election-iomgr-refactor) */ diff --git a/src/include/typemap.h b/src/include/typemap.h index 61d1cb4..5e340b0 100644 --- a/src/include/typemap.h +++ b/src/include/typemap.h @@ -17,6 +17,8 @@ typedef struct upgrade_config_req_s upgrade_config_req_t; typedef struct heartbeat_msg_s heartbeat_msg_t; typedef struct ctrl_hello_req_s ctrl_hello_req_t; typedef struct ctrl_hello_reply_s ctrl_hello_reply_t; +typedef struct wal_catchup_entry wal_catchup_entry_t; +typedef struct snapshot_transfer_s snapshot_transfer_t; typedef struct { char* resname; @@ -192,6 +194,15 @@ typedef struct { uint64_t config_tag; } state_report_t; +/* WAL apply ACK — sent by standby to primary after applying a WAL catch-up + * entry (Task 3.1, ha-wal-replication). Must be declared before xce_reply_t + * so the walackreplymsg union arm compiles correctly. */ +typedef struct { + uint64_t term; + uint64_t lsn; + int success; +} wal_ack_reply_t; + typedef struct { uint64_t seq; uint32_t opcode; @@ -208,6 +219,7 @@ typedef struct { ctrl_hello_reply_t* ctrlhelloreplymsg; request_vote_reply_t* requestvotereplymsg; state_report_t* state_report; + wal_ack_reply_t* walackreplymsg; } data; /* Req 7: standby redirect fields — populated when status == XCE_ERR_STANDBY */ char *redirect_host; @@ -244,8 +256,8 @@ typedef struct { heartbeat_msg_t* heartbeatmsg; ctrl_hello_req_t* ctrlhellomsg; request_vote_req_t* requestvotemsg; - void* walcatchupentrymsg; /* wal_catchup_entry_t* */ - void* snapshottransfermsg; /* snapshot_transfer_t* */ + wal_catchup_entry_t* walcatchupentrymsg; + snapshot_transfer_t* snapshottransfermsg; } body; } xce_request_t; @@ -269,34 +281,43 @@ typedef struct { uint64_t lsn; } wal_ack_req_t; -typedef struct { - uint64_t term; - uint64_t lsn; - int success; -} wal_ack_reply_t; - typedef struct { uint64_t term; uint64_t acked_tag; int success; } upgrade_config_ack_t; -typedef struct { +struct snapshot_transfer_s { uint64_t snapshot_lsn; /* checkpoint LSN this snapshot corresponds to */ - xce_data_t *payload; /* chunk payload (64 KB max) */ + xce_data_t *payload; /* legacy decoded field — broken for raw bytes + * (decode_Data_ptr misinterprets List(UInt8)); + * kept NULL on send, do not use */ uint32_t chunk_index; /* 0-based chunk number */ uint64_t chunk_offset; /* byte offset in the full snapshot */ uint64_t total_size; /* total snapshot size in bytes */ int is_last; /* true (non-zero) for the final chunk */ -} snapshot_transfer_t; + /* Raw chunk bytes — same pattern as repl_config_push_req_t. + * Send side: filled by the sender, encoded by xce_snapshot_chunk_req_raw. + * Receive side: extracted by xce_decode() before capn_free(); malloc'd, + * owned by the struct, freed by xce_free_request(). */ + uint8_t *raw_payload; + size_t raw_payload_len; +}; -typedef struct wal_catchup_entry { +struct wal_catchup_entry { uint64_t lsn; uint8_t entry_type; - xce_data_t *payload; + xce_data_t *payload; /* legacy decoded field — broken for + * raw bytes; kept NULL, do not use */ int n_entries; /* batch: number of sub-entries */ struct wal_catchup_entry **entries; /* batch: array of sub-entry pointers */ -} wal_catchup_entry_t; + /* Raw WAL entry payload bytes — same pattern as repl_config_push_req_t. + * Send side: filled from the WAL segment file, encoded per sub-entry by + * xce_wal_batch_req_raw. Receive side: extracted by xce_decode() before + * capn_free(); malloc'd, owned, freed by xce_free_request(). */ + uint8_t *raw_payload; + size_t raw_payload_len; +}; /* ── Persistence Cap'n Proto mapped types (P0-9) ───────────────────── * These are the C-side counterparts for the capnp encode/decode shims. diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 23f9838..dbf4c5e 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -560,6 +560,7 @@ struct RequestReply $C.mapname("xce_reply_t"){ ctrlhelloreplymsg @13: CtrlHelloReply; requestvotereplymsg @14: RequestVoteReply; stateReportMsg @15: StateReport $C.mapname("state_report"); # heartbeat reply: standby state + walackreplymsg @16: WalAckReply $C.mapname("walackreplymsg"); # WAL apply ACK from standby } redirectHost @11: Text $C.mapname("redirect_host"); @@ -626,6 +627,8 @@ struct Request $C.mapname("xce_request_t"){ heartbeatmsg @25: HeartbeatMsg; ctrlhellomsg @26: CtrlHelloRequest; requestvotemsg @27: RequestVoteRequest; + walcatchupentrymsg @28: WalCatchupEntry; + snapshottransfermsg @29: SnapshotTransfer; } }