From 8bfa861be2405db8903101d6f34a3e55d2cd2661 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 20 Jun 2026 11:52:18 +0000 Subject: [PATCH] feat(ha-heartbeat-committed-propagation): add reconcile-free ha_observe_heartbeat_committed (task 2.1) Strict-monotonic updater for a standby's learned committed position: advance last_known_committed_lsn only when strictly greater (zero/lower no-op), set the config tag together (tag-before-lsn release ordering for reader safety). No ha_rejoin_reconcile, no truncation, no other mutation (Req 2.1-2.3, 3.1, 3.3). 4 revert-detecting unit tests; controller suite 0 failures. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../tasks.md | 2 +- src/controller/ha.c | 40 ++++ src/controller/ha.h | 21 ++ src/controller/meson.build | 1 + src/controller/testsuite/controller_test.c | 2 + .../testsuite/test_ha_heartbeat_committed.c | 208 ++++++++++++++++++ 6 files changed, 273 insertions(+), 1 deletion(-) create mode 100644 src/controller/testsuite/test_ha_heartbeat_committed.c diff --git a/.kiro/specs/ha-heartbeat-committed-propagation/tasks.md b/.kiro/specs/ha-heartbeat-committed-propagation/tasks.md index cb627c45..040b0d0a 100644 --- a/.kiro/specs/ha-heartbeat-committed-propagation/tasks.md +++ b/.kiro/specs/ha-heartbeat-committed-propagation/tasks.md @@ -17,7 +17,7 @@ - _Boundary: libxce sender_ - [ ] 2. Standby consume (reconcile-free, term-guarded, monotonic) -- [ ] 2.1 Add the reconcile-free monotonic updater `ha_observe_heartbeat_committed` +- [x] 2.1 Add the reconcile-free monotonic updater `ha_observe_heartbeat_committed` - Add `ha_observe_heartbeat_committed(cluster_t* c, uint64_t committed_lsn, uint64_t committed_config_tag)` (`ha.c`, declared in `ha.h`): advance `last_known_committed_lsn` only when the carried LSN is strictly greater (monotonic; atomic), advancing `last_known_committed_config_tag` together with it; no-op when `<=` current or zero. It must NOT call `ha_rejoin_reconcile`, truncate the WAL, or modify any produced/persisted committed state. - TDD: unit tests assert advance-on-higher (LSN + config tag updated), no-op on equal/lower/zero, config tag never regresses independently, and — critically — that the call performs no WAL truncation / no reconcile and mutates nothing beyond `last_known_committed_*`. - Observable: the unit tests go red → green; the no-truncation / no-mutation assertions hold. diff --git a/src/controller/ha.c b/src/controller/ha.c index 4feb65d2..6cf1dccd 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -4919,6 +4919,46 @@ void ha_on_heartbeat_received(ha_state_t *ha) } } +/* + * ha_observe_heartbeat_committed: reconcile-free monotonic learn of the + * primary's committed position from a heartbeat (ha-heartbeat-committed- + * propagation, Req 2.1, 2.2, 2.3, 3.1, 3.3). + * + * Single-receiver-thread path (controller request handling), so a relaxed + * compare-then-store is sufficient. The config tag is written BEFORE the LSN + * (both release-ordered) so a concurrent acquire-load reader that observes the + * advanced LSN never pairs it with a stale tag. + * + * Advisory only: NO ha_rejoin_reconcile / NO WAL truncation / NO change to any + * produced/persisted committed, role, term, or applied state. This is the + * explicit distinction from cluster_apply_ctrlhello_reply (which truncates). + */ +void ha_observe_heartbeat_committed(cluster_t *c, uint64_t committed_lsn, + uint64_t committed_config_tag) +{ + if (!c) return; + ha_state_t *ha = &c->ha; + + uint64_t cur = atomic_load_explicit(&ha->last_known_committed_lsn, + memory_order_acquire); + /* Monotonic no-op: a zero (absent/liveness-only) or non-advancing LSN must + * leave the learned position — and its config tag — untouched (Req 2.2). */ + if (committed_lsn == 0 || committed_lsn <= cur) + return; + + /* Tag moves with the LSN, never independently regressed (Req 2.3): publish + * the tag first so the LSN's release store carries the matching tag. */ + atomic_store_explicit(&ha->last_known_committed_config_tag, + committed_config_tag, memory_order_release); + atomic_store_explicit(&ha->last_known_committed_lsn, committed_lsn, + memory_order_release); + /* Reconcile-free: nothing else is touched (Req 3.1, 3.3). */ + tlog(TLOG_DEBUG3, + "HA: learned committed_lsn=%" PRIu64 " config_tag=%" PRIu64 + " from heartbeat (was %" PRIu64 ")", + committed_lsn, committed_config_tag, cur); +} + void ha_disarm_heartbeat_timeout_vtask(ha_state_t *ha) { (void)ha; diff --git a/src/controller/ha.h b/src/controller/ha.h index a98506e1..9392c4f1 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -878,6 +878,27 @@ void ha_discover_primary(ha_state_t *ha, cluster_t *c); /* ha_on_heartbeat_received: reset the heartbeat deadline (Req 6.3). */ void ha_on_heartbeat_received(ha_state_t *ha); +/* + * ha_observe_heartbeat_committed: reconcile-free monotonic update of a standby's + * learned committed position from a primary heartbeat + * (ha-heartbeat-committed-propagation, Req 2.1, 2.2, 2.3, 3.1, 3.3). + * + * Advances ha->last_known_committed_lsn to committed_lsn ONLY when it is + * strictly greater than the current value (monotonic); when it advances, the + * carried committed_config_tag is stored together with it (they move as a pair — + * the tag never regresses independently). No-op when committed_lsn is 0 or + * <= the current learned LSN. + * + * This is the reconcile-free sibling of cluster_apply_ctrlhello_reply: it MUST + * NOT call ha_rejoin_reconcile, truncate the WAL, or modify ANY other state + * (produced/persisted committed, role, term, applied, etc.). The value is + * advisory — for observability and vote-floor freshness only. The caller (the + * OP_nop receive path) applies the term-guard before invoking this; the helper + * itself is term-agnostic and single-receiver-thread safe. + */ +void ha_observe_heartbeat_committed(cluster_t *c, uint64_t committed_lsn, + uint64_t committed_config_tag); + /* ha_heartbeat_is_fresh: 1 if a primary message arrived within * heartbeat_timeout_ms (leader-stickiness predicate). Exposed for PreVote * denial-reason logging in cluster.c; pure read, no state change. */ diff --git a/src/controller/meson.build b/src/controller/meson.build index 35261bbb..edd4e5b5 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -134,6 +134,7 @@ controller_unit_test_src = [ 'testsuite/test_coldstart_lsn0.c', 'testsuite/test_commit_durability.c', 'testsuite/test_ctrlstatus.c', + 'testsuite/test_ha_heartbeat_committed.c', 'ctrlstatus.c', 'commit.c', 'job.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 2f48718b..0ebfad3f 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -61,6 +61,7 @@ extern void test_convergence_verification(void); extern void test_coldstart_lsn0(void); extern void test_commit_durability(void); extern void test_ctrlstatus(void); +extern void test_ha_heartbeat_committed(void); void setUp(void) { return; } @@ -191,6 +192,7 @@ int main(int argc, char* argv[]) TEST(test_coldstart_lsn0); TEST(test_commit_durability); TEST(test_ctrlstatus); + TEST(test_ha_heartbeat_committed); return UNITY_END(); } diff --git a/src/controller/testsuite/test_ha_heartbeat_committed.c b/src/controller/testsuite/test_ha_heartbeat_committed.c new file mode 100644 index 00000000..99cb0863 --- /dev/null +++ b/src/controller/testsuite/test_ha_heartbeat_committed.c @@ -0,0 +1,208 @@ +/* test_ha_heartbeat_committed.c — unit tests for the reconcile-free monotonic + * heartbeat-borne committed updater (ha-heartbeat-committed-propagation, + * task 2.1). + * + * Exercises ha_observe_heartbeat_committed(cluster_t*, committed_lsn, + * committed_config_tag): the standby-side consumer that monotonically advances + * ha->last_known_committed_lsn / last_known_committed_config_tag from a primary + * heartbeat WITHOUT any reconciliation or WAL truncation. This is the explicit + * reconcile-free sibling of cluster_apply_ctrlhello_reply (which DOES truncate). + * + * Covers Requirements 2.1, 2.2, 2.3, 3.1, 3.3. + */ +#include +#include +#include +#include + +#include + +#include "ha.h" +#include "xce/cluster_def.h" /* full cluster_t (embedded ha_state_t / persist_t) */ + +/* ── helpers ────────────────────────────────────────────────── */ + +static ha_config_t hb_valid_cfg(void) +{ + 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, + }; + return cfg; +} + +/* Build a minimal cluster_t whose ha state the updater touches. Only the fields + * ha_observe_heartbeat_committed() reads/writes are initialised by ha_init. */ +static void hb_setup_cluster(cluster_t* c) +{ + memset(c, 0, sizeof(*c)); + ha_config_t cfg = hb_valid_cfg(); + TEST_ASSERT_EQUAL_INT(0, ha_init(&c->ha, &cfg, 0)); +} + +static void hb_teardown_cluster(cluster_t* c) +{ + ha_fini(&c->ha); +} + +static uint64_t hb_lsn(const cluster_t* c) +{ + return atomic_load_explicit(&c->ha.last_known_committed_lsn, + memory_order_acquire); +} + +static uint64_t hb_tag(const cluster_t* c) +{ + return atomic_load_explicit(&c->ha.last_known_committed_config_tag, + memory_order_acquire); +} + +/* ── tests ──────────────────────────────────────────────────── */ + +/* (a) Advance on a strictly higher carried LSN: both the LSN and the config tag + * move together to the carried values. (Req 2.1, 2.3) */ +static void test_observe_advances_on_higher(void) +{ + cluster_t c; + hb_setup_cluster(&c); + + atomic_store_explicit(&c.ha.last_known_committed_lsn, 4, + memory_order_release); + atomic_store_explicit(&c.ha.last_known_committed_config_tag, 0x11, + memory_order_release); + + ha_observe_heartbeat_committed(&c, 9, 0xAA); + + TEST_ASSERT_EQUAL_UINT64(9, hb_lsn(&c)); /* advanced (2.1) */ + TEST_ASSERT_EQUAL_UINT64(0xAA, hb_tag(&c)); /* tag moved with lsn (2.3) */ + + hb_teardown_cluster(&c); + printf("test_observe_advances_on_higher: PASS\n"); +} + +/* (b) No-op on equal or lower carried LSN: neither the LSN nor the tag changes, + * proving the config tag never regresses independently of the LSN. (Req 2.2, 2.3) */ +static void test_observe_noop_on_equal_or_lower(void) +{ + cluster_t c; + hb_setup_cluster(&c); + + atomic_store_explicit(&c.ha.last_known_committed_lsn, 9, + memory_order_release); + atomic_store_explicit(&c.ha.last_known_committed_config_tag, 0xAA, + memory_order_release); + + /* equal LSN, different tag → no change (2.2): tag must NOT take 0xBB */ + ha_observe_heartbeat_committed(&c, 9, 0xBB); + TEST_ASSERT_EQUAL_UINT64(9, hb_lsn(&c)); + TEST_ASSERT_EQUAL_UINT64(0xAA, hb_tag(&c)); + + /* lower LSN (out-of-order / lagging primary), different tag → no change */ + ha_observe_heartbeat_committed(&c, 3, 0xCC); + TEST_ASSERT_EQUAL_UINT64(9, hb_lsn(&c)); /* unchanged (2.2) */ + TEST_ASSERT_EQUAL_UINT64(0xAA, hb_tag(&c)); /* no independent regress (2.3) */ + + hb_teardown_cluster(&c); + printf("test_observe_noop_on_equal_or_lower: PASS\n"); +} + +/* (c) No-op on a zero carried LSN: an absent/liveness-only committed value must + * leave the known position untouched. (Req 2.2 / liveness no-op) */ +static void test_observe_noop_on_zero(void) +{ + cluster_t c; + hb_setup_cluster(&c); + + atomic_store_explicit(&c.ha.last_known_committed_lsn, 5, + memory_order_release); + atomic_store_explicit(&c.ha.last_known_committed_config_tag, 0x55, + memory_order_release); + + ha_observe_heartbeat_committed(&c, 0, 0xDD); + + TEST_ASSERT_EQUAL_UINT64(5, hb_lsn(&c)); /* unchanged */ + TEST_ASSERT_EQUAL_UINT64(0x55, hb_tag(&c)); /* unchanged */ + + hb_teardown_cluster(&c); + printf("test_observe_noop_on_zero: PASS\n"); +} + +/* (d) Reconcile-free: an advancing observe must mutate ONLY last_known_committed_* + * and nothing else — no WAL truncation, no produced/persisted committed change, + * no role/term/applied change. (Req 3.1, 3.3) */ +static void test_observe_no_reconcile_no_other_mutation(void) +{ + cluster_t c; + hb_setup_cluster(&c); + + /* Seed the authoritative / produced state the helper must NOT touch. */ + atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY, + memory_order_release); + atomic_store_explicit(&c.ha.term, 7, memory_order_release); + atomic_store_explicit(&c.ha.applied_lsn, 100, memory_order_release); + atomic_store_explicit(&c.persist.committed_lsn, 80, memory_order_release); + atomic_store_explicit(&c.persist.durable_lsn, 80, memory_order_release); + c.persist.current_lsn = 81; /* WAL tip (non-atomic, under p->lock in prod) */ + atomic_store_explicit(&c.ha.committed_config_tag, 0xC0FFEE, + memory_order_release); + atomic_store_explicit(&c.ha.last_known_committed_lsn, 60, + memory_order_release); + atomic_store_explicit(&c.ha.last_known_committed_config_tag, 0x6060, + memory_order_release); + + /* Snapshot the produced/persisted state that MUST remain untouched. */ + int32_t role0 = atomic_load_explicit(&c.ha.role, memory_order_acquire); + uint64_t term0 = atomic_load_explicit(&c.ha.term, memory_order_acquire); + uint64_t applied0 = atomic_load_explicit(&c.ha.applied_lsn, + memory_order_acquire); + uint64_t pcommit0 = atomic_load_explicit(&c.persist.committed_lsn, + memory_order_acquire); + uint64_t pdur0 = atomic_load_explicit(&c.persist.durable_lsn, + memory_order_acquire); + uint64_t ptip0 = c.persist.current_lsn; + uint64_t ctag0 = atomic_load_explicit(&c.ha.committed_config_tag, + memory_order_acquire); + + /* Advance the learned committed via the heartbeat path. */ + ha_observe_heartbeat_committed(&c, 75, 0x7575); + + /* ONLY last_known_committed_* changed. */ + TEST_ASSERT_EQUAL_UINT64(75, hb_lsn(&c)); + TEST_ASSERT_EQUAL_UINT64(0x7575, hb_tag(&c)); + + /* No truncation: the persisted/produced committed tip is intact (3.1, 3.3). */ + TEST_ASSERT_EQUAL_UINT64(pcommit0, + atomic_load_explicit(&c.persist.committed_lsn, + memory_order_acquire)); + TEST_ASSERT_EQUAL_UINT64(pdur0, + atomic_load_explicit(&c.persist.durable_lsn, + memory_order_acquire)); + TEST_ASSERT_EQUAL_UINT64(ptip0, c.persist.current_lsn); + /* No other HA state moved. */ + TEST_ASSERT_EQUAL_INT(role0, + atomic_load_explicit(&c.ha.role, memory_order_acquire)); + TEST_ASSERT_EQUAL_UINT64(term0, + atomic_load_explicit(&c.ha.term, memory_order_acquire)); + TEST_ASSERT_EQUAL_UINT64(applied0, + atomic_load_explicit(&c.ha.applied_lsn, + memory_order_acquire)); + TEST_ASSERT_EQUAL_UINT64(ctag0, + atomic_load_explicit(&c.ha.committed_config_tag, + memory_order_acquire)); + + hb_teardown_cluster(&c); + printf("test_observe_no_reconcile_no_other_mutation: PASS\n"); +} + +/* ── suite entry point ─────────────────────────────────────── */ + +void test_ha_heartbeat_committed(void) +{ + RUN_TEST(test_observe_advances_on_higher); + RUN_TEST(test_observe_noop_on_equal_or_lower); + RUN_TEST(test_observe_noop_on_zero); + RUN_TEST(test_observe_no_reconcile_no_other_mutation); +}