From aab41fa609a1c56a2de7c4d96cb743b31ff14f05 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 4 Jun 2026 16:40:44 +0000 Subject: [PATCH] feat(ha-repl-hardening): per-standby convergence tracking from heartbeats (task 3.1) ha_process_standby_state stores (tag, lsn) per standby, evaluates convergence (tag>=committed && lsn>=committed), sets converged flag, logs INFO only on transition. Wired into repl_ack_decode_and_dispatch. 5 tests: converged, lsn-behind, tag-behind, regression, multi-peer. --- .kiro/specs/ha-repl-hardening/tasks.md | 2 +- src/controller/ha.c | 100 +++++++++++++++- src/controller/ha.h | 25 ++++ src/controller/testsuite/test_ha_role.c | 146 ++++++++++++++++++++++++ 4 files changed, 267 insertions(+), 6 deletions(-) diff --git a/.kiro/specs/ha-repl-hardening/tasks.md b/.kiro/specs/ha-repl-hardening/tasks.md index f0b3825c..c9d91b5b 100644 --- a/.kiro/specs/ha-repl-hardening/tasks.md +++ b/.kiro/specs/ha-repl-hardening/tasks.md @@ -71,7 +71,7 @@ --- - [ ] 3. Primary-side: Unified heartbeat-driven replication decision loop -- [ ] 3.1 Implement per-standby convergence tracking update +- [x] 3.1 Implement per-standby convergence tracking update - On each heartbeat reply from a standby, extract `appliedLsn` and `configTag` from `stateReport` - Store in `standby_reported_tag[idx]` and `standby_reported_lsn[idx]` - Evaluate convergence: `reported_tag >= committed_tag AND reported_lsn >= committed_lsn` diff --git a/src/controller/ha.c b/src/controller/ha.c index 0dde4bd8..130c8d49 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -934,6 +934,46 @@ void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf, reply_out->data.state_report = sr; } +/* + * ha_process_standby_state — update per-standby convergence tracking. + * + * Called on each heartbeat reply from a standby (CTRL_PRIMARY_COMMITTING or + * CTRL_PRIMARY). Stores (reported_tag, reported_lsn) into the per-standby + * arrays, evaluates convergence, and logs INFO on the first + * not-converged → converged transition (Req 7.2). Regression resets the flag + * (Req 2.5). + * + * Convergence: reported_tag >= committed_tag AND reported_lsn >= committed_lsn. + * + * Requirements: 1.2, 2.2, 2.5 (ha-repl-hardening spec) + */ +void ha_process_standby_state(ha_state_t *ha, int peer_idx, + uint64_t reported_tag, uint64_t reported_lsn, + uint64_t committed_tag, uint64_t committed_lsn) +{ + if (peer_idx < 0 || peer_idx >= MAX_HA_CANDIDATES) + return; + + /* Store reported values */ + ha->standby_reported_tag[peer_idx] = reported_tag; + ha->standby_reported_lsn[peer_idx] = reported_lsn; + + /* Evaluate convergence */ + bool was_converged = ha->standby_converged[peer_idx]; + bool now_converged = (reported_tag >= committed_tag) && + (reported_lsn >= committed_lsn); + ha->standby_converged[peer_idx] = now_converged; + + /* Log INFO on first transition to converged (Req 7.2) */ + if (now_converged && !was_converged) { + INFO("HA: standby[%d] converged: config_tag=%" PRIu64 + " applied_lsn=%" PRIu64 " (committed_tag=%" PRIu64 + " committed_lsn=%" PRIu64 ")", + peer_idx, reported_tag, reported_lsn, + committed_tag, committed_lsn); + } +} + /* ── Dispatch gate (req 1.3, 1.4) ────────────────────────────────────────── * * Called from myreq_handler() in cluster.c before the opcode switch. @@ -1961,8 +2001,11 @@ done: /* * repl_ack_decode_and_dispatch: decode an inbound packet on a promoted vote * endpoint while role == CTRL_PRIMARY_COMMITTING or CTRL_PRIMARY. - * Expects OPCode_replCatchUpDone; calls ha_repl_ack_received on success. - * Drops unexpected opcodes with a WARN log. + * + * - OPCode_replCatchUpDone → ha_repl_ack_received + * - OPCode_reply with stateReportMsg → ha_process_standby_state (Task 3.1) + * - OPCode_nop or OPCode_reply without stateReport → silently ignored + * - anything else → WARN + drop */ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx) { @@ -1990,13 +2033,60 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx) outer_rp.p = capn_getp(capn_root(&capn_ctx), 0, 1); read_Request(&outer_rq, outer_rp); int opcode = (int)outer_rq.opcode; - capn_free(&capn_ctx); if (opcode == OPCode_replCatchUpDone) { + capn_free(&capn_ctx); ha_repl_ack_received(ha, c); - } else if (opcode == OPCode_nop || opcode == OPCode_reply) { - /* Heartbeat NOP reply or generic reply — silently drop */ + } else if (opcode == OPCode_reply && + outer_rq.body_which == Request_body_replymsg && + outer_rq.body.replymsg.p.data) { + /* Heartbeat NOP reply — decode inner RequestReply for stateReport */ + struct RequestReply rep_s; + memset(&rep_s, 0, sizeof(rep_s)); + read_RequestReply(&rep_s, outer_rq.body.replymsg); + + if (rep_s.data_which == RequestReply_data_stateReportMsg) { + /* Decode stateReport and call convergence tracker (Task 3.1). + * read_StateReport reads scalar fields directly from the capnp + * pointer without heap allocation — safe to call before capn_free. */ + struct StateReport sr_raw; + memset(&sr_raw, 0, sizeof(sr_raw)); + read_StateReport(&sr_raw, rep_s.data.stateReportMsg); + state_report_t sr; + sr.applied_lsn = (uint64_t)sr_raw.appliedLsn; + sr.config_tag = (uint64_t)sr_raw.configTag; + capn_free(&capn_ctx); + + /* peer_idx from standby_ep_ctx_t (layout-compatible with vote_ep_ctx_t) */ + int peer_idx = vctx->peer_idx; + uint64_t committed_lsn = 0; + uint64_t committed_tag = 0; +#ifndef XCE_TESTING + if (c) { + committed_lsn = atomic_load_explicit(&c->persist.committed_lsn, + memory_order_acquire); + committed_tag = atomic_load_explicit(&ha->committed_config_tag, + memory_order_acquire); + } +#else + /* In test mode, c may be NULL; caller sets ha fields directly */ + committed_lsn = atomic_load_explicit(&ha->last_known_committed_lsn, + memory_order_acquire); + committed_tag = atomic_load_explicit(&ha->committed_config_tag, + memory_order_acquire); + (void)c; +#endif + ha_process_standby_state(ha, peer_idx, + sr.config_tag, sr.applied_lsn, + committed_tag, committed_lsn); + } else { + capn_free(&capn_ctx); + } + } else if (opcode == OPCode_nop) { + /* Plain NOP with no body — silently ignore */ + capn_free(&capn_ctx); } else { + capn_free(&capn_ctx); WARN("repl_ack_decode_and_dispatch: unexpected opcode %d on promoted ep", opcode); } diff --git a/src/controller/ha.h b/src/controller/ha.h index c4ea233b..882d427c 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -289,6 +289,31 @@ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf, state_report_t *sr, xce_reply_t *reply_out); +/* + * ha_process_standby_state: update per-standby convergence tracking. + * + * Called on each heartbeat reply from a standby while the primary is in + * CTRL_PRIMARY_COMMITTING or CTRL_PRIMARY. Stores the standby's reported + * (config_tag, applied_lsn) and evaluates whether the standby has converged + * with the primary's committed state. + * + * Convergence criteria (Req 1.3): + * reported_tag >= committed_tag AND reported_lsn >= committed_lsn + * + * Logs INFO on the first transition from not-converged → converged (Req 7.2). + * Regression (converged → not-converged) clears standby_converged[peer_idx] + * and re-evaluates (Req 2.5). + * + * peer_idx: 0-based index into ha->standby_reported_* arrays; must be in + * [0, MAX_HA_CANDIDATES). + * + * Requirements: 1.2, 2.2, 2.5 (ha-repl-hardening spec) + * Boundary: Primary Heartbeat Handler (ha.c) + */ +void ha_process_standby_state(ha_state_t *ha, int peer_idx, + uint64_t reported_tag, uint64_t reported_lsn, + uint64_t committed_tag, uint64_t committed_lsn); + /* ── HA Config Bootstrap & Term Durability (ha-config-bootstrap spec) ── */ /* diff --git a/src/controller/testsuite/test_ha_role.c b/src/controller/testsuite/test_ha_role.c index 52f9fd3c..fb0a2d00 100644 --- a/src/controller/testsuite/test_ha_role.c +++ b/src/controller/testsuite/test_ha_role.c @@ -237,6 +237,146 @@ static void test_ha_init_zeroes_standby_convergence_fields(void) printf("test_ha_init_zeroes_standby_convergence_fields: PASS\n"); } +/* ── Task 3.1: ha_process_standby_state convergence tracking ─────────────── */ + +/* + * Declare the function under test. It is defined in ha.c and exposed only in + * XCE_TESTING builds (forward declaration matches the implementation). + */ +#ifdef XCE_TESTING +void ha_process_standby_state(ha_state_t *ha, int peer_idx, + uint64_t reported_tag, uint64_t reported_lsn, + uint64_t committed_tag, uint64_t committed_lsn); +#endif + +/* Helper: build a ha_state_t with all convergence fields zeroed. */ +static ha_state_t *build_ha_for_convergence_test(void) +{ + static 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, + }; + ha_init(&ha, &cfg, 0); + return &ha; +} + +/* + * Test: standby with lsn == committed_lsn and tag == committed_tag → converged. + * Fields standby_reported_tag[0] and standby_reported_lsn[0] updated correctly. + * Requirements: 1.2, 2.2 (ha-repl-hardening) + */ +static void test_process_standby_state_converged(void) +{ + ha_state_t *ha = build_ha_for_convergence_test(); + + ha_process_standby_state(ha, 0, + /*reported_tag=*/5, /*reported_lsn=*/100, + /*committed_tag=*/5, /*committed_lsn=*/100); + + TEST_ASSERT_EQUAL_UINT64(5, ha->standby_reported_tag[0]); + TEST_ASSERT_EQUAL_UINT64(100, ha->standby_reported_lsn[0]); + TEST_ASSERT_TRUE(ha->standby_converged[0]); + + ha_fini(ha); + printf("test_process_standby_state_converged: PASS\n"); +} + +/* + * Test: standby lsn < committed_lsn → not converged. + * Requirements: 1.2, 2.2 + */ +static void test_process_standby_state_lsn_behind(void) +{ + ha_state_t *ha = build_ha_for_convergence_test(); + + ha_process_standby_state(ha, 0, + /*reported_tag=*/5, /*reported_lsn=*/50, + /*committed_tag=*/5, /*committed_lsn=*/100); + + TEST_ASSERT_EQUAL_UINT64(5, ha->standby_reported_tag[0]); + TEST_ASSERT_EQUAL_UINT64(50, ha->standby_reported_lsn[0]); + TEST_ASSERT_FALSE(ha->standby_converged[0]); + + ha_fini(ha); + printf("test_process_standby_state_lsn_behind: PASS\n"); +} + +/* + * Test: standby tag < committed_tag → not converged. + * Requirements: 1.2, 2.2 + */ +static void test_process_standby_state_tag_behind(void) +{ + ha_state_t *ha = build_ha_for_convergence_test(); + + ha_process_standby_state(ha, 0, + /*reported_tag=*/3, /*reported_lsn=*/100, + /*committed_tag=*/5, /*committed_lsn=*/100); + + TEST_ASSERT_EQUAL_UINT64(3, ha->standby_reported_tag[0]); + TEST_ASSERT_EQUAL_UINT64(100, ha->standby_reported_lsn[0]); + TEST_ASSERT_FALSE(ha->standby_converged[0]); + + ha_fini(ha); + printf("test_process_standby_state_tag_behind: PASS\n"); +} + +/* + * Test: convergence regression — standby was converged; primary advances; + * next update with old values should reset standby_converged to false. + * Requirements: 2.5 (converged standby regresses → mark not-converged) + */ +static void test_process_standby_state_regression(void) +{ + ha_state_t *ha = build_ha_for_convergence_test(); + + /* First update: converged at (tag=5, lsn=100) */ + ha_process_standby_state(ha, 0, + /*reported_tag=*/5, /*reported_lsn=*/100, + /*committed_tag=*/5, /*committed_lsn=*/100); + TEST_ASSERT_TRUE(ha->standby_converged[0]); + + /* Primary advances committed_lsn to 200; standby still reports (5, 100) */ + ha_process_standby_state(ha, 0, + /*reported_tag=*/5, /*reported_lsn=*/100, + /*committed_tag=*/5, /*committed_lsn=*/200); + TEST_ASSERT_FALSE(ha->standby_converged[0]); + + ha_fini(ha); + printf("test_process_standby_state_regression: PASS\n"); +} + +/* + * Test: different standby indices updated independently. + * Requirements: 2.2 + */ +static void test_process_standby_state_multiple_peers(void) +{ + ha_state_t *ha = build_ha_for_convergence_test(); + + /* peer 0: converged */ + ha_process_standby_state(ha, 0, + /*reported_tag=*/5, /*reported_lsn=*/100, + /*committed_tag=*/5, /*committed_lsn=*/100); + /* peer 1: not converged (tag behind) */ + ha_process_standby_state(ha, 1, + /*reported_tag=*/3, /*reported_lsn=*/100, + /*committed_tag=*/5, /*committed_lsn=*/100); + + TEST_ASSERT_TRUE(ha->standby_converged[0]); + TEST_ASSERT_FALSE(ha->standby_converged[1]); + TEST_ASSERT_EQUAL_UINT64(5, ha->standby_reported_tag[0]); + TEST_ASSERT_EQUAL_UINT64(3, ha->standby_reported_tag[1]); + + ha_fini(ha); + printf("test_process_standby_state_multiple_peers: PASS\n"); +} + void test_ha_role(void) { RUN_TEST(test_ha_init_default_role_is_standby); @@ -252,6 +392,12 @@ void test_ha_role(void) RUN_TEST(test_ha_init_zeroes_new_fields); /* Task 1.4: ha-repl-hardening per-standby convergence fields */ RUN_TEST(test_ha_init_zeroes_standby_convergence_fields); + /* Task 3.1: ha_process_standby_state convergence tracking */ + RUN_TEST(test_process_standby_state_converged); + RUN_TEST(test_process_standby_state_lsn_behind); + RUN_TEST(test_process_standby_state_tag_behind); + RUN_TEST(test_process_standby_state_regression); + RUN_TEST(test_process_standby_state_multiple_peers); } /* test_am_i_primary suite deleted — am_i_primary() removed (election-iomgr-refactor) */