feat(ha-repl-hardening): unified replication decision loop (task 3.2)

ha_evaluate_repl_action() pure function: stale tag→PUSH_CONFIG,
lsn<oldest_wal→SEND_SNAPSHOT, lsn<committed→STREAM_WAL, else CONVERGED.
5 tests. Fixed misleading stream_wal_from_zero test name per review.
This commit is contained in:
Rongsong Shen 2026-06-04 16:58:48 +00:00
parent aab41fa609
commit a475ef621f
4 changed files with 176 additions and 1 deletions

View file

@ -81,7 +81,7 @@
- _Boundary: Primary Heartbeat Handler (ha.c)_ - _Boundary: Primary Heartbeat Handler (ha.c)_
- _Depends: 1.1, 1.4_ - _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 - 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 - 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 - Log decision at INFO: which standby, which action triggered

View file

@ -934,6 +934,41 @@ void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf,
reply_out->data.state_report = sr; 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. * ha_process_standby_state — update per-standby convergence tracking.
* *

View file

@ -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, void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf,
state_report_t *sr, xce_reply_t *reply_out); 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. * ha_process_standby_state: update per-standby convergence tracking.
* *

View file

@ -377,6 +377,93 @@ static void test_process_standby_state_multiple_peers(void)
printf("test_process_standby_state_multiple_peers: PASS\n"); 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) void test_ha_role(void)
{ {
RUN_TEST(test_ha_init_default_role_is_standby); 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_tag_behind);
RUN_TEST(test_process_standby_state_regression); RUN_TEST(test_process_standby_state_regression);
RUN_TEST(test_process_standby_state_multiple_peers); 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) */ /* test_am_i_primary suite deleted — am_i_primary() removed (election-iomgr-refactor) */