From 7dc657b1bde7ea66fee5dccc913af55f1b3ef14e Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 4 Jun 2026 16:08:11 +0000 Subject: [PATCH] feat(ha-repl-hardening): populate stateReport in standby heartbeat reply (task 2.3) OPCode_nop handler now sends stateReport {applied_lsn, config_tag} instead of nodata. Function in ha.c for unit-test linkability. Two tests: warm state and zero-state fresh standby (Req 1.4). --- .kiro/specs/ha-repl-hardening/tasks.md | 2 +- src/controller/cluster.c | 12 ++- src/controller/ha.c | 16 ++++ src/controller/ha.h | 11 ++- src/controller/testsuite/test_config_repl.c | 92 +++++++++++++++++++++ 5 files changed, 128 insertions(+), 5 deletions(-) diff --git a/.kiro/specs/ha-repl-hardening/tasks.md b/.kiro/specs/ha-repl-hardening/tasks.md index 1abf0738..f0b3825c 100644 --- a/.kiro/specs/ha-repl-hardening/tasks.md +++ b/.kiro/specs/ha-repl-hardening/tasks.md @@ -60,7 +60,7 @@ - _Boundary: Standby Dispatch (cluster.c / ha.c)_ - _Depends: 1.2, 2.1_ -- [ ] 2.3 (P) Populate `stateReport` in heartbeat reply on standby +- [x] 2.3 (P) Populate `stateReport` in heartbeat reply on standby - In the `OPCode_nop` handler (`cluster.c`), replace `data_which = nodata` with `data_which = stateReport` - Populate `appliedLsn` from `ha->applied_lsn` and `configTag` from `myconf->config_version_tag` - Unit test: verify reply carries correct `applied_lsn` and `config_tag` after WAL catch-up diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 2e897c45..e9706d67 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -172,6 +172,7 @@ static int handle_wal_catchup_entry(cluster_t *c, xce_request_t *req) return 0; } + static int myreq_handler(vtask_t* task) { vtask_data_t* data = (vtask_data_t*)task->data; @@ -240,9 +241,13 @@ static int myreq_handler(vtask_t* task) xce_reply(ep, c->myhost, &reply); } break; - case OPCode_nop: - reply.status = 0; - reply.data_which = RequestReply_data_nodata; + case OPCode_nop: { + /* Task 2.3: standby populates stateReport in every heartbeat reply so + * the primary can track convergence (Req 1.1, 1.3, 1.4). */ + state_report_t sr; + memset(&sr, 0, sizeof(sr)); + reply.status = 0; + 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. */ @@ -250,6 +255,7 @@ static int myreq_handler(vtask_t* task) == CTRL_STANDBY) ha_on_heartbeat_received(&c->ha); break; + } case OPCode_join: rc = handle_join(c, ep, &request); break; diff --git a/src/controller/ha.c b/src/controller/ha.c index c5e8a4f3..0dde4bd8 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -918,6 +918,22 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, return 0; } +/* ha_nop_reply_state_report — populate *sr from ha state and wire into reply. + * Reads ha->applied_lsn (atomic) and myconf->config_version_tag, stores them + * in *sr, then sets reply_out->data_which = stateReportMsg and + * reply_out->data.state_report = sr. + * Caller keeps *sr alive until xce_reply() serialises the message. + * Requirements: 1.1, 1.3, 1.4 (ha-repl-hardening spec) */ +void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf, + state_report_t *sr, xce_reply_t *reply_out) +{ + sr->applied_lsn = atomic_load_explicit(&ha->applied_lsn, + memory_order_acquire); + sr->config_tag = myconf ? myconf->config_version_tag : 0; + reply_out->data_which = RequestReply_data_stateReportMsg; + reply_out->data.state_report = sr; +} + /* ── Dispatch gate (req 1.3, 1.4) ────────────────────────────────────────── * * Called from myreq_handler() in cluster.c before the opcode switch. diff --git a/src/controller/ha.h b/src/controller/ha.h index b25b3bf1..c4ea233b 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -26,8 +26,9 @@ #include #include "config.h" /* XCE_MAXNAME_LEN, MAX_STANDBYS, etc. */ +#include "myconf.h" /* myconf_t, config_version_tag */ #include "persist.h" /* persist_t, apply_wal_entry */ -#include "typemap.h" /* wal_catchup_entry_t */ +#include "typemap.h" /* wal_catchup_entry_t, state_report_t, xce_reply_t */ #include "xce/cluster.h" /* xce_candidate_t */ #include "zarray.h" #include "raft/raft.h" /* raft_state_t, raft_init */ @@ -280,6 +281,14 @@ int ha_dispatch_gate(ha_state_t *ha, int opcode); int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, wal_catchup_entry_t *entry); +/* ha_nop_reply_state_report — populate a heartbeat reply with stateReport. + * Fills *sr from ha->applied_lsn (atomic read) and myconf->config_version_tag, + * then sets reply_out->data_which = stateReportMsg and + * reply_out->data.state_report = sr. + * Caller must keep *sr live until xce_reply() serialises the reply. */ +void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf, + state_report_t *sr, xce_reply_t *reply_out); + /* ── HA Config Bootstrap & Term Durability (ha-config-bootstrap spec) ── */ /* diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index 337659dd..2a8253fd 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -1238,6 +1238,95 @@ static void test_wal_catchup_batch_zero_entries_fallback(void) printf("test_wal_catchup_batch_zero_entries_fallback: PASS\n"); } +/* ── Task 2.3: OPCode_nop heartbeat reply carries stateReport ──────────── */ + +#ifdef XCE_TESTING +/* Test seam declared in cluster.c under #ifdef XCE_TESTING — + * fills *sr from ha->applied_lsn / myconf->config_version_tag, + * sets reply_out->data_which = stateReportMsg and reply_out->data.state_report = sr. + * Req 1.1, 1.3, 1.4. */ +void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf, + state_report_t *sr, xce_reply_t *reply_out); + +/* Test 2.3-A: nop reply carries stateReport with applied_lsn and config_tag. + * Requirements 1.1, 1.3, 1.4: standby populates stateReport in every + * OPCode_nop heartbeat reply with current applied_lsn and config_version_tag. */ +static void test_nop_reply_carries_state_report(void) +{ + ha_state_t ha; + memset(&ha, 0, sizeof(ha)); + ha_config_t cfg; + memset(&cfg, 0, sizeof(cfg)); + cfg.heartbeat_timeout_ms = 10000; + cfg.election_timeout_ms = 1500; + cfg.committing_timeout_ms = 10000; + ha_init(&ha, &cfg, 0); + + /* Set a known applied_lsn */ + atomic_store_explicit(&ha.applied_lsn, 12345ULL, memory_order_release); + + /* Build a minimal myconf with a known config_version_tag */ + myconf_t myconf; + memset(&myconf, 0, sizeof(myconf)); + myconf.config_version_tag = 0xABCD1234ULL; + + state_report_t sr; + memset(&sr, 0, sizeof(sr)); + xce_reply_t reply; + memset(&reply, 0, sizeof(reply)); + + ha_nop_reply_state_report(&ha, &myconf, &sr, &reply); + + /* Reply must use stateReportMsg discriminant */ + TEST_ASSERT_EQUAL_INT(RequestReply_data_stateReportMsg, reply.data_which); + + /* state_report pointer must point at our sr */ + TEST_ASSERT_EQUAL_PTR(&sr, reply.data.state_report); + + /* applied_lsn must match ha->applied_lsn */ + TEST_ASSERT_EQUAL_UINT64(12345ULL, sr.applied_lsn); + + /* config_tag must match myconf->config_version_tag */ + TEST_ASSERT_EQUAL_UINT64(0xABCD1234ULL, sr.config_tag); + + ha_fini(&ha); + printf("test_nop_reply_carries_state_report: PASS\n"); +} + +/* Test 2.3-B: nop reply when applied_lsn==0 (fresh standby, no config). + * Requirement 1.4: config_tag reports zero when no config received yet. */ +static void test_nop_reply_zero_state_fresh_standby(void) +{ + ha_state_t ha; + memset(&ha, 0, sizeof(ha)); + ha_config_t cfg; + memset(&cfg, 0, sizeof(cfg)); + cfg.heartbeat_timeout_ms = 10000; + cfg.election_timeout_ms = 1500; + cfg.committing_timeout_ms = 10000; + ha_init(&ha, &cfg, 0); + + /* applied_lsn stays at 0; config_version_tag stays at 0 → no config yet */ + myconf_t myconf; + memset(&myconf, 0, sizeof(myconf)); + + state_report_t sr; + memset(&sr, 0, sizeof(sr)); + xce_reply_t reply; + memset(&reply, 0, sizeof(reply)); + + ha_nop_reply_state_report(&ha, &myconf, &sr, &reply); + + TEST_ASSERT_EQUAL_INT(RequestReply_data_stateReportMsg, reply.data_which); + TEST_ASSERT_EQUAL_PTR(&sr, reply.data.state_report); + TEST_ASSERT_EQUAL_UINT64(0ULL, sr.applied_lsn); + TEST_ASSERT_EQUAL_UINT64(0ULL, sr.config_tag); + + ha_fini(&ha); + printf("test_nop_reply_zero_state_fresh_standby: PASS\n"); +} +#endif /* XCE_TESTING */ + #endif /* XCE_TESTING */ /* ── dispatch ───────────────────────────────────── */ @@ -1275,6 +1364,9 @@ void test_config_repl(void) RUN_TEST(test_wal_catchup_partial_batch); RUN_TEST(test_wal_catchup_batch_error_entry_skipped); RUN_TEST(test_wal_catchup_batch_zero_entries_fallback); + /* Task 2.3: OPCode_nop heartbeat reply carries stateReport */ + RUN_TEST(test_nop_reply_carries_state_report); + RUN_TEST(test_nop_reply_zero_state_fresh_standby); #endif /* Task A: raw capnp shims */ RUN_TEST(test_replconfig_req_ssl_encode);