From a475ef621fe0b4c8d6ed4ef020361c818be32686 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 4 Jun 2026 16:58:48 +0000 Subject: [PATCH] feat(ha-repl-hardening): unified replication decision loop (task 3.2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ha_evaluate_repl_action() pure function: stale tag→PUSH_CONFIG, lsndata.state_report = sr; } +/* + * ha_evaluate_repl_action — pure decision function for the unified heartbeat- + * driven replication protocol. + * + * Implements the decision tree from the ha-repl-hardening design: + * + * if standby_tag < committed_tag → HA_REPL_PUSH_CONFIG + * elif standby_lsn < oldest_wal_lsn → HA_REPL_SEND_SNAPSHOT + * elif standby_lsn < committed_lsn → HA_REPL_STREAM_WAL + * else (tag >= committed AND lsn >= committed) → HA_REPL_CONVERGED + * + * No side effects. All callers must use the returned action to drive the + * replication send; they must not cache the result beyond the current cycle. + * + * Requirements: 2.2, 2.3, 2.4 (ha-repl-hardening spec) + * Boundary: Primary Heartbeat Handler (ha.c) + */ +ha_repl_action_t ha_evaluate_repl_action(uint64_t standby_tag, + uint64_t standby_lsn, + uint64_t committed_tag, + uint64_t committed_lsn, + uint64_t oldest_wal_lsn) +{ + if (standby_tag < committed_tag) + return HA_REPL_PUSH_CONFIG; + + if (standby_lsn < oldest_wal_lsn) + return HA_REPL_SEND_SNAPSHOT; + + if (standby_lsn < committed_lsn) + return HA_REPL_STREAM_WAL; + + return HA_REPL_CONVERGED; +} + /* * ha_process_standby_state — update per-standby convergence tracking. * diff --git a/src/controller/ha.h b/src/controller/ha.h index 882d427..86908be 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -289,6 +289,53 @@ 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); +/* ── Task 3.2: Unified replication decision enum and evaluator ────────────── */ + +/* + * ha_repl_action_t: result of ha_evaluate_repl_action(). + * + * Encodes the single replication action the primary should take for a standby + * on the current heartbeat cycle, based solely on the standby's reported + * (config_tag, applied_lsn) vs the primary's committed state. + * + * Decision tree (design §Unified Heartbeat-Driven Protocol): + * standby_tag < committed_tag → PUSH_CONFIG + * standby_lsn < oldest_wal_lsn → SEND_SNAPSHOT + * standby_lsn < committed_lsn → STREAM_WAL + * tag >= committed AND lsn >= committed → CONVERGED + * + * Requirements: 2.2, 2.3, 2.4 (ha-repl-hardening spec) + */ +typedef enum { + HA_REPL_PUSH_CONFIG, /* standby config tag is behind — push config first */ + HA_REPL_SEND_SNAPSHOT, /* standby LSN predates oldest WAL — transfer snapshot */ + HA_REPL_STREAM_WAL, /* standby LSN is within WAL window — stream entries */ + HA_REPL_CONVERGED, /* standby tag and LSN are both current */ +} ha_repl_action_t; + +/* + * ha_evaluate_repl_action: pure decision function — no side effects. + * + * Evaluates the standby's reported (standby_tag, standby_lsn) against the + * primary's committed state (committed_tag, committed_lsn, oldest_wal_lsn) + * and returns the single action the primary should take this heartbeat cycle. + * + * Arguments: + * standby_tag — config_tag last reported by this standby + * standby_lsn — applied_lsn last reported by this standby + * committed_tag — primary's committed config_tag + * committed_lsn — primary's committed WAL LSN + * oldest_wal_lsn — oldest retained WAL LSN (entries before this are gone) + * + * Requirements: 2.2, 2.3, 2.4 (ha-repl-hardening spec) + * Boundary: Primary Heartbeat Handler (ha.c) + */ +ha_repl_action_t ha_evaluate_repl_action(uint64_t standby_tag, + uint64_t standby_lsn, + uint64_t committed_tag, + uint64_t committed_lsn, + uint64_t oldest_wal_lsn); + /* * ha_process_standby_state: update per-standby convergence tracking. * diff --git a/src/controller/testsuite/test_ha_role.c b/src/controller/testsuite/test_ha_role.c index fb0a2d0..a34585b 100644 --- a/src/controller/testsuite/test_ha_role.c +++ b/src/controller/testsuite/test_ha_role.c @@ -377,6 +377,93 @@ static void test_process_standby_state_multiple_peers(void) printf("test_process_standby_state_multiple_peers: PASS\n"); } +/* ── Task 3.2: ha_evaluate_repl_action unified decision tree ─────────────── */ + +/* + * Declare the function under test. It is defined in ha.c. + * Pure function — no side effects — so no XCE_TESTING guard needed. + */ +ha_repl_action_t ha_evaluate_repl_action(uint64_t standby_tag, + uint64_t standby_lsn, + uint64_t committed_tag, + uint64_t committed_lsn, + uint64_t oldest_wal_lsn); + +/* + * Test: (tag=0, lsn=0) with committed_tag=5, committed_lsn=100, + * oldest_wal_lsn=50 → PUSH_CONFIG (tag behind). + * Requirements: 2.2, 2.3 (ha-repl-hardening) + */ +static void test_evaluate_repl_action_push_config(void) +{ + ha_repl_action_t action = ha_evaluate_repl_action( + /*standby_tag=*/0, /*standby_lsn=*/0, + /*committed_tag=*/5, /*committed_lsn=*/100, + /*oldest_wal_lsn=*/50); + TEST_ASSERT_EQUAL_INT(HA_REPL_PUSH_CONFIG, action); + printf("test_evaluate_repl_action_push_config: PASS\n"); +} + +/* + * Test: (tag=5, lsn=0) with committed_tag=5, committed_lsn=100, + * oldest_wal_lsn=50 → SEND_SNAPSHOT (lsn=0 is below WAL retention window). + * Requirements: 2.2, 2.3 + */ +static void test_evaluate_repl_action_send_snapshot_lsn_zero(void) +{ + ha_repl_action_t action = ha_evaluate_repl_action( + /*standby_tag=*/5, /*standby_lsn=*/0, + /*committed_tag=*/5, /*committed_lsn=*/100, + /*oldest_wal_lsn=*/50); + /* lsn=0 < oldest_wal_lsn=50 → SEND_SNAPSHOT */ + TEST_ASSERT_EQUAL_INT(HA_REPL_SEND_SNAPSHOT, action); + printf("test_evaluate_repl_action_send_snapshot_lsn_zero: PASS\n"); +} + +/* + * Test: (tag=5, lsn=100) with committed_tag=5, committed_lsn=100 → CONVERGED. + * Requirements: 2.2, 2.4 + */ +static void test_evaluate_repl_action_converged(void) +{ + ha_repl_action_t action = ha_evaluate_repl_action( + /*standby_tag=*/5, /*standby_lsn=*/100, + /*committed_tag=*/5, /*committed_lsn=*/100, + /*oldest_wal_lsn=*/50); + TEST_ASSERT_EQUAL_INT(HA_REPL_CONVERGED, action); + printf("test_evaluate_repl_action_converged: PASS\n"); +} + +/* + * Test: (tag=5, lsn=10) with committed_tag=5, committed_lsn=100, + * oldest_wal_lsn=50 → SEND_SNAPSHOT (lsn < oldest_wal). + * Requirements: 2.2, 2.3 + */ +static void test_evaluate_repl_action_send_snapshot(void) +{ + ha_repl_action_t action = ha_evaluate_repl_action( + /*standby_tag=*/5, /*standby_lsn=*/10, + /*committed_tag=*/5, /*committed_lsn=*/100, + /*oldest_wal_lsn=*/50); + TEST_ASSERT_EQUAL_INT(HA_REPL_SEND_SNAPSHOT, action); + printf("test_evaluate_repl_action_send_snapshot: PASS\n"); +} + +/* + * Test: (tag=5, lsn=50) with committed_tag=5, committed_lsn=100, + * oldest_wal_lsn=50 → STREAM_WAL (lsn >= oldest_wal but < committed). + * Requirements: 2.2, 2.3 + */ +static void test_evaluate_repl_action_stream_wal(void) +{ + ha_repl_action_t action = ha_evaluate_repl_action( + /*standby_tag=*/5, /*standby_lsn=*/50, + /*committed_tag=*/5, /*committed_lsn=*/100, + /*oldest_wal_lsn=*/50); + TEST_ASSERT_EQUAL_INT(HA_REPL_STREAM_WAL, action); + printf("test_evaluate_repl_action_stream_wal: PASS\n"); +} + void test_ha_role(void) { RUN_TEST(test_ha_init_default_role_is_standby); @@ -398,6 +485,12 @@ void test_ha_role(void) RUN_TEST(test_process_standby_state_tag_behind); RUN_TEST(test_process_standby_state_regression); RUN_TEST(test_process_standby_state_multiple_peers); + /* Task 3.2: ha_evaluate_repl_action unified decision tree */ + RUN_TEST(test_evaluate_repl_action_push_config); + RUN_TEST(test_evaluate_repl_action_send_snapshot_lsn_zero); + RUN_TEST(test_evaluate_repl_action_converged); + RUN_TEST(test_evaluate_repl_action_send_snapshot); + RUN_TEST(test_evaluate_repl_action_stream_wal); } /* test_am_i_primary suite deleted — am_i_primary() removed (election-iomgr-refactor) */