From d534f83e460b2e37e060cca946d895d52b3c9e18 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 20 Jun 2026 12:22:50 +0000 Subject: [PATCH] feat(ha-heartbeat-committed-propagation): stamp own committed+term on primary heartbeat (task 3.1) primary_heartbeat_fn now sends xce_ha_heartbeat_req with the primary's OWN atomic ha->term / persist.committed_lsn / committed_config_tag (NOT last_known_*), snapshotted once per tick (Req 1.1, 3.4). XCE_TESTING seam adds a capturing ha_heartbeat_req_hook (precedence new->legacy->real) so existing ha_nop_req_hook liveness tests stay green; send-failure WARN + last_push_time unchanged. New test poisons last_known_* to prove own-committed. Direct binary 119/0. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../tasks.md | 4 +- src/controller/ha.c | 43 +++++++- src/controller/ha.h | 16 +++ src/controller/testsuite/test_ha_liveness.c | 98 +++++++++++++++++++ 4 files changed, 158 insertions(+), 3 deletions(-) diff --git a/.kiro/specs/ha-heartbeat-committed-propagation/tasks.md b/.kiro/specs/ha-heartbeat-committed-propagation/tasks.md index 125d51e6..af4d2791 100644 --- a/.kiro/specs/ha-heartbeat-committed-propagation/tasks.md +++ b/.kiro/specs/ha-heartbeat-committed-propagation/tasks.md @@ -33,7 +33,7 @@ - _Depends: 2.1_ - [ ] 3. Primary produce + safety verification -- [ ] 3.1 Stamp committed + term on the primary heartbeat +- [x] 3.1 Stamp committed + term on the primary heartbeat - In `primary_heartbeat_fn` (`ha.c`), replace the per-standby `xce_nop_req(ep)` heartbeat call with `xce_ha_heartbeat_req(ep, atomic_load(ha->term), atomic_load(persist.committed_lsn), atomic_load(ha->committed_config_tag))`. Preserve the `XCE_TESTING` `ha_nop_req_hook` seam and the existing heartbeat send-failure `WARN` and `last_push_time` bookkeeping. - TDD/observable: a test (or the `ha_nop_req_hook` seam) confirms the primary's heartbeat now carries its current committed LSN / committed config tag / term; the send-failure and liveness-eviction behavior of `primary_heartbeat_fn` is unchanged. - _Requirements: 1.1, 3.4_ @@ -58,5 +58,7 @@ - Do NOT reuse `cluster_apply_ctrlhello_reply` for the heartbeat path — it truncates (`ha_rejoin_reconcile`). The heartbeat updater is reconcile-free (Req 3.1). - Do NOT touch the client keepalive `OPCode_heartbeat` / `xce_heartbeat` (a different message) or the generic `xce_nop_req` (client nop). - 1.1 LEARNING: `ha.h` is `src/controller/ha.h` (NOT `src/controller/include/ha.h`). `last_known_committed_config_tag` is now `_Atomic uint64_t` (`ha.h:172`), sibling of `last_known_committed_lsn` (`ha.h:170`); read via `atomic_load_explicit(acquire)`, write via `atomic_store_explicit(release)`. +- TEST-RUN LEARNING (2.x/3.x): the meson test wrapper (`xce-build.sh controller`) is NON-DETERMINISTIC ("OK"/"FAIL"/"Timeout") because of ~149 PRE-EXISTING LeakSanitizer leaks from `ha_start_election_async`/`ha_start_prevote_async` (`test_ha_start_election_async.c`, `test_ha_rdhup_integration.c`) — unrelated to this spec (identical count on a stashed base). AUTHORITATIVE verification = run the binary directly: `cd src && ASAN_OPTIONS=detect_leaks=0 ./xce.build/controller/controller_unit_test`, then check rc=0 and zero `:FAIL:` markers. +- 3.1 LEARNING: primary stamps its OWN `ha->term` / `c->persist.committed_lsn` / `ha->committed_config_tag` (all `_Atomic`, acquire-load once per tick) — NOT the standby `last_known_*`. Test seam: new `ha_heartbeat_req_hook(ep,term,lsn,tag)` (`XCE_TESTING`, NULL default) with precedence new-hook → legacy `ha_nop_req_hook` → real `xce_ha_heartbeat_req`, so existing `ha_nop_req_hook` liveness tests stay green. - 1.2 LEARNING: the C-side `heartbeatmsg` body support PRE-EXISTED (no typemap edit needed) — `heartbeat_msg_t` typedef `typemap.h:17`, `xce_request_t.heartbeatmsg` member `typemap.h:287`, `struct heartbeat_msg_s{sender_priority,term,committed_lsn,committed_config_tag}` `typemap.h:445` (order matches schema `@0..@3`); enum `Request_body_heartbeatmsg == 22` (rides `OPCode_nop`). The codec (`encode/decode/free_HeartbeatMsg` + the request arm) was generated; only producer/consumer were missing. Test round-trip pattern: heap-`calloc` packet consumed+freed by `xce_decode` (caller does NOT free pkt), decoded body freed by `xce_free_request` (see `test_wal_wire.c`). Sender uses `xce_request(ep, &req)` (non-blocking). - The vote-floor formula is NOT edited — task 3.2 is a verification test only. Safety rests on the term-guard (task 2.2) + monotonicity (task 2.1) + cluster-monotonic committed; see design.md Security Considerations. diff --git a/src/controller/ha.c b/src/controller/ha.c index 47d948c7..9e3b5e47 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -2764,6 +2764,17 @@ int (*ha_push_config_hook)(void *c) = NULL; */ int (*ha_nop_req_hook)(endpoint_t *ep) = NULL; +/* + * ha_heartbeat_req_hook: test-only seam for the HA heartbeat send in + * primary_heartbeat_fn. When non-NULL, primary_heartbeat_fn calls this INSTEAD + * OF xce_ha_heartbeat_req with the SAME (term, committed_lsn, + * committed_config_tag) it stamps onto the wire, so tests can assert the + * heartbeat carries the primary's own current committed state. Req 1.1, 3.4. + */ +int (*ha_heartbeat_req_hook)(endpoint_t *ep, uint64_t term, + uint64_t committed_lsn, + uint64_t committed_config_tag) = NULL; + /* * ha_repl_push_config_for_standby_hook: test-only seam for the per-standby * config push triggered by the heartbeat-reply decision loop (Task 3.3). @@ -5286,6 +5297,20 @@ static int primary_heartbeat_fn(void *arg) if (timeout_s == 0) timeout_s = HA_DEFAULT_HEARTBEAT_TIMEOUT_MS / 1000; + /* Stamp the primary's OWN current committed position + term onto + * every heartbeat this tick (Req 1.1, 3.4). Read atomically here, + * once per tick, so every standby sees a consistent snapshot. Use + * the primary's authoritative state — NOT the standby-side + * last_known_* fields. */ + uint64_t hb_term = + atomic_load_explicit(&ha->term, memory_order_acquire); + uint64_t hb_committed_lsn = + atomic_load_explicit(&c->persist.committed_lsn, + memory_order_acquire); + uint64_t hb_committed_config_tag = + atomic_load_explicit(&ha->committed_config_tag, + memory_order_acquire); + /* Endpoints evicted this tick; their repl deregistration runs AFTER * ha->lock is released (repl lock-ordering — mirrors the err handlers). */ endpoint_t *evicted[MAX_HA_CANDIDATES]; @@ -5324,10 +5349,24 @@ static int primary_heartbeat_fn(void *arg) time_t elapsed = now - sctx->last_push_time; if (elapsed >= (time_t)interval_s) { + /* Heartbeat carries the primary's own committed + term + * (Req 1.1, 3.4). */ #ifdef XCE_TESTING - int rc = ha_nop_req_hook ? ha_nop_req_hook(ep) : xce_nop_req(ep); + int rc; + if (ha_heartbeat_req_hook) + rc = ha_heartbeat_req_hook(ep, hb_term, + hb_committed_lsn, + hb_committed_config_tag); + else if (ha_nop_req_hook) + rc = ha_nop_req_hook(ep); + else + rc = xce_ha_heartbeat_req(ep, hb_term, + hb_committed_lsn, + hb_committed_config_tag); #else - int rc = xce_nop_req(ep); + int rc = xce_ha_heartbeat_req(ep, hb_term, + hb_committed_lsn, + hb_committed_config_tag); #endif if (rc < 0) WARN("HA: heartbeat nop to standby[%d] failed: %d", i, rc); diff --git a/src/controller/ha.h b/src/controller/ha.h index 19de6fa6..e06892d1 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -1076,6 +1076,22 @@ extern int (*ha_send_ctrlhello_hook)(void *c); */ extern int (*ha_push_config_hook)(void *c); +/* + * ha_heartbeat_req_hook: test-only seam for the per-standby HA heartbeat send + * in primary_heartbeat_fn (Task 3.1). When non-NULL, primary_heartbeat_fn + * calls this INSTEAD OF xce_ha_heartbeat_req, passing the SAME three values it + * stamps onto the wire — the primary's CURRENT own term / committed_lsn / + * committed_config_tag. A test installs it to capture those args and assert + * the heartbeat carries the primary's own committed state (Req 1.1, 3.4). + * + * Precedence (XCE_TESTING only): ha_heartbeat_req_hook (if set) takes priority, + * else the legacy ha_nop_req_hook (if set), else the real xce_ha_heartbeat_req. + * Keeping ha_nop_req_hook as a fallback preserves the existing liveness tests. + */ +extern int (*ha_heartbeat_req_hook)(endpoint_t *ep, uint64_t term, + uint64_t committed_lsn, + uint64_t committed_config_tag); + /* * ha_repl_push_config_for_standby_hook: test-only seam for the per-standby * config push triggered by the heartbeat-reply decision loop (Task 3.3). diff --git a/src/controller/testsuite/test_ha_liveness.c b/src/controller/testsuite/test_ha_liveness.c index d16da9de..ee2df4eb 100644 --- a/src/controller/testsuite/test_ha_liveness.c +++ b/src/controller/testsuite/test_ha_liveness.c @@ -313,6 +313,17 @@ typedef struct { */ extern int (*ha_nop_req_hook)(endpoint_t *ep); +/* + * ha_heartbeat_req_hook: XCE_TESTING hook (task 3.1). When non-NULL, + * primary_heartbeat_fn calls this with the (term, committed_lsn, + * committed_config_tag) it stamps onto the heartbeat, so a test can assert the + * heartbeat carries the primary's OWN current committed state. Declared in + * ha.h. + */ +extern int (*ha_heartbeat_req_hook)(endpoint_t *ep, uint64_t term, + uint64_t committed_lsn, + uint64_t committed_config_tag); + /* * primary_heartbeat_fn_for_test: test entry-point that exercises the primary * heartbeat idle-tracking logic directly without a live delaymgr. @@ -340,6 +351,24 @@ static int stub_nop_req(endpoint_t *ep) return 0; } +/* Capture for ha_heartbeat_req_hook (task 3.1): the args the primary stamps. */ +static int g_hb_calls = 0; +static uint64_t g_hb_term = 0; +static uint64_t g_hb_committed_lsn = 0; +static uint64_t g_hb_committed_config_tag = 0; + +static int stub_heartbeat_req(endpoint_t *ep, uint64_t term, + uint64_t committed_lsn, + uint64_t committed_config_tag) +{ + (void)ep; + g_hb_calls++; + g_hb_term = term; + g_hb_committed_lsn = committed_lsn; + g_hb_committed_config_tag = committed_config_tag; + return 0; +} + static void init_ha_primary_with_standby_eps(ha_state_t *ha) { ha_init(ha, &g_cfg, 0); @@ -546,6 +575,74 @@ static void test_primary_hb_keeps_unconfirmed_standby(void) printf("test_primary_hb_keeps_unconfirmed_standby: PASS\n"); } +/* ── Test 9e: primary stamps its OWN current committed + term on heartbeat ── + * + * Task 3.1 / Req 1.1, 3.4: when an idle standby is due a heartbeat, the primary + * must send xce_ha_heartbeat_req carrying its CURRENT OWN state read atomically: + * term = ha->term + * committed_lsn = c->persist.committed_lsn + * committed_config_tag = ha->committed_config_tag + * NOT the standby-side last_known_* fields. The ha_heartbeat_req_hook captures + * those three args; we assert they equal the primary's own values. + */ +static void test_primary_hb_stamps_own_committed_and_term(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + init_ha_primary_with_standby_eps(&c.ha); + + /* Primary's own authoritative state — distinct sentinel values so a wrong + * source (e.g. last_known_* or a hardcoded 0) is caught. */ + const uint64_t PRIMARY_TERM = 7; + const uint64_t PRIMARY_COMMITTED_LSN = 42; + const uint64_t PRIMARY_COMMITTED_TAG = 13; + atomic_store_explicit(&c.ha.term, PRIMARY_TERM, memory_order_release); + atomic_store_explicit(&c.persist.committed_lsn, PRIMARY_COMMITTED_LSN, + memory_order_release); + atomic_store_explicit(&c.ha.committed_config_tag, PRIMARY_COMMITTED_TAG, + memory_order_release); + + /* Poison the standby-side last_known_* fields with DIFFERENT values: a + * heartbeat that mistakenly read these (the 3.4 bug) would carry them. */ + atomic_store_explicit(&c.ha.last_known_committed_lsn, 999, + memory_order_release); + atomic_store_explicit(&c.ha.last_known_committed_config_tag, 888, + memory_order_release); + + /* Idle standby (last_push_time long ago) so the heartbeat is due. */ + endpoint_t ep; + memset(&ep, 0, sizeof(ep)); + standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0); + standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data; + sctx->last_push_time = 0; + zarray_store(c.ha.standby_endpoints, &ep); + + g_hb_calls = 0; + g_hb_term = g_hb_committed_lsn = g_hb_committed_config_tag = 0; + ha_heartbeat_req_hook = stub_heartbeat_req; + + primary_heartbeat_fn_for_test(&c); + + ha_heartbeat_req_hook = NULL; + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_hb_calls, + "Req 1.1: primary must send a heartbeat (xce_ha_heartbeat_req) to an idle standby"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(PRIMARY_TERM, g_hb_term, + "Req 3.4: heartbeat must carry the primary's OWN current term"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(PRIMARY_COMMITTED_LSN, g_hb_committed_lsn, + "Req 1.1/3.4: heartbeat must carry the primary's OWN persist.committed_lsn"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(PRIMARY_COMMITTED_TAG, g_hb_committed_config_tag, + "Req 1.1/3.4: heartbeat must carry the primary's OWN committed_config_tag"); + + /* last_push_time still updated on success — bookkeeping unchanged. */ + TEST_ASSERT_NOT_EQUAL_MESSAGE(0, (int)sctx->last_push_time, + "last_push_time must be updated after a successful heartbeat send"); + + free(ep.data); + fini_ha_primary_with_standby_eps(&c.ha); + printf("test_primary_hb_stamps_own_committed_and_term: PASS\n"); +} + /* ── Test 10: on_standby_connected sets last_push_time ──────────────────── */ /* @@ -762,6 +859,7 @@ void test_ha_liveness(void) RUN_TEST(test_primary_hb_evicts_silent_standby); RUN_TEST(test_primary_hb_keeps_live_standby); RUN_TEST(test_primary_hb_keeps_unconfirmed_standby); + RUN_TEST(test_primary_hb_stamps_own_committed_and_term); RUN_TEST(test_on_standby_connected_sets_last_push_time); RUN_TEST(test_cluster_dispatch_nop_resets_activity_when_standby); RUN_TEST(test_cluster_dispatch_nop_skips_when_primary);