diff --git a/.kiro/specs/ha-repl-hardening/tasks.md b/.kiro/specs/ha-repl-hardening/tasks.md index c9d91b5b..363f0d34 100644 --- a/.kiro/specs/ha-repl-hardening/tasks.md +++ b/.kiro/specs/ha-repl-hardening/tasks.md @@ -81,7 +81,7 @@ - _Boundary: Primary Heartbeat Handler (ha.c)_ - _Depends: 1.1, 1.4_ -- [ ] 3.2 Implement unified decision loop in heartbeat reply processing +- [x] 3.2 Implement unified decision loop in heartbeat reply processing - After updating per-standby tracking, evaluate: tag stale → push config; `lsn < oldest_wal_lsn` → send snapshot; `lsn < committed_lsn` → stream WAL; both current → converged - One action per heartbeat cycle per standby; `stream_wal_entries()` enqueues all batches non-blocking via `xce_request()` in a single invocation — never blocks vtaskmgr thread - Log decision at INFO: which standby, which action triggered diff --git a/src/controller/ha.c b/src/controller/ha.c index 130c8d49..8ac6ebbb 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -934,6 +934,41 @@ void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf, reply_out->data.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 882d427c..86908be1 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 fb0a2d00..a34585b5 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) */