lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(ha-heartbeat-committed-propagation): stamp own committed+term on primary heartbeat (task 3.1)

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-06-20 12:22:50 +00:00
parent 90e7179a89
commit d534f83e46
4 changed files with 158 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

43
src/controller/ha.c
View file

@ -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);

16
src/controller/ha.h
View file

@ -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).

98
src/controller/testsuite/test_ha_liveness.c
View file

@ -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);