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).
This commit is contained in:
Rongsong Shen 2026-06-04 16:08:11 +00:00
parent 6bbcff62ca
commit 7dc657b1bd
5 changed files with 128 additions and 5 deletions

View file

@ -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

View file

@ -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;

View file

@ -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.

View file

@ -26,8 +26,9 @@
#include <time.h>
#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) ── */
/*

View file

@ -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);