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

feat(controller-status): carry election time on the heartbeat (task 7.3)

Browse source 
The primary echoes its own last_known_elected_at_ms on every outgoing
heartbeat (threaded through xce_ha_heartbeat_req's existing scalar-arg
plumbing, alongside term/committed_lsn/committed_config_tag); a standby
processing a received heartbeat (ha_consume_heartbeat_committed, the sole
production call site) stores it unconditionally into its own
last_known_elected_at_ms, reconcile-free -- no other HA field moves.
ha_observe_heartbeat_committed itself is untouched; the new store sits
next to its call site, not inside it.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RLSNJhNoULvVcUhXEFafQC
This commit is contained in:
Rongsong Shen 2026-09-16 06:03:03 +00:00
parent 21ae00b375
commit 30adc52ea3
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
9 changed files with 252 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

166
src/controller/testsuite/test_ha_heartbeat_committed.c
View file

@ -61,6 +61,12 @@ static uint64_t hb_tag(const cluster_t* c)
memory_order_acquire);
}
static uint64_t hb_elected(const cluster_t* c)
{
return atomic_load_explicit(&c->ha.last_known_elected_at_ms,
memory_order_acquire);
}
/* ── tests ──────────────────────────────────────────────────── */
/* (a) Advance on a strictly higher carried LSN: both the LSN and the config tag
@ -332,6 +338,162 @@ static void test_consume_null_or_zero_noop(void)
printf("test_consume_null_or_zero_noop: PASS\n");
}
/* ── controller-status delta (Req 12.2): learn the primary's election time
* off a received heartbeat ─────────────────────────────────────────────────
*
* ha_consume_heartbeat_committed's SAME call site that feeds
* ha_observe_heartbeat_committed(c, hb->committed_lsn, hb->committed_config_tag)
* also carries hb->elected_at_ms — a genuinely-RECEIVED heartbeat (the
* production caller is cluster.c's OP_nop dispatch) already gates on
* non-primary role + a non-stale term before either call, so the
* elected_at_ms store is a plain unconditional overwrite immediately
* alongside the existing committed-position learn — no separate monotonic/
* term compare (design.md Locked Decision 2). ha_observe_heartbeat_committed
* itself is untouched; these tests exercise ONLY the new sibling store.
*/
/* (i) A term-valid heartbeat stores the carried election time as the
* standby's own learned value (Req 12.2). */
static void test_consume_stores_elected_at_ms(void)
{
cluster_t c;
hb_setup_cluster(&c);
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY,
memory_order_release);
atomic_store_explicit(&c.ha.term, 5, memory_order_release);
atomic_store_explicit(&c.ha.last_known_elected_at_ms, 111,
memory_order_release);
heartbeat_msg_t hb = {
.sender_priority = 0,
.term = 5, /* == standby term -> passes term-guard */
.committed_lsn = 9,
.committed_config_tag = 0xAA,
.elected_at_ms = 222222,
};
ha_consume_heartbeat_committed(&c, &hb);
TEST_ASSERT_EQUAL_UINT64(222222, hb_elected(&c)); /* learned (12.2) */
hb_teardown_cluster(&c);
printf("test_consume_stores_elected_at_ms: PASS\n");
}
/* (j) Reconcile-free: a heartbeat carrying ONLY a new election time (its
* committed_lsn is 0, a liveness-only no-op for the committed observe) moves
* last_known_elected_at_ms and NOTHING else -- committed position, role,
* term, and applied position are all untouched. Mirrors
* test_observe_no_reconcile_no_other_mutation's shape for the new field
* (Req 12.2, 12.5). This is the RED test: it fails (elected_at_ms stays at
* its seeded value) before the store exists in ha_consume_heartbeat_committed. */
static void test_consume_elected_at_ms_reconcile_free(void)
{
cluster_t c;
hb_setup_cluster(&c);
/* Seed every learned/produced HA field the store must NOT touch. */
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY,
memory_order_release);
atomic_store_explicit(&c.ha.term, 5, 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;
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);
atomic_store_explicit(&c.ha.last_known_elected_at_ms, 1111,
memory_order_release);
/* Snapshot everything that must remain byte-for-byte unchanged. */
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);
uint64_t klsn0 = hb_lsn(&c);
uint64_t ktag0 = hb_tag(&c);
heartbeat_msg_t hb = {
.sender_priority = 0,
.term = 5, /* == standby term -> passes term-guard */
.committed_lsn = 0, /* liveness-only -> observe no-ops (4.2) */
.committed_config_tag = 0xDEAD,
.elected_at_ms = 9999,
};
ha_consume_heartbeat_committed(&c, &hb);
/* ONLY last_known_elected_at_ms changed. */
TEST_ASSERT_EQUAL_UINT64(9999, hb_elected(&c));
/* Committed position -- untouched (observe no-op on committed_lsn==0). */
TEST_ASSERT_EQUAL_UINT64(klsn0, hb_lsn(&c));
TEST_ASSERT_EQUAL_UINT64(ktag0, hb_tag(&c));
/* No truncation / no produced-state mutation. */
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_consume_elected_at_ms_reconcile_free: PASS\n");
}
/* (k) A stale-term heartbeat (hb->term < ha->term) must NOT store its carried
* election time either -- the elected_at_ms store sits behind the SAME
* term-guard early-return as the committed-position learn (3.2 shape). */
static void test_consume_stale_term_does_not_store_elected_at_ms(void)
{
cluster_t c;
hb_setup_cluster(&c);
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.last_known_elected_at_ms, 4242,
memory_order_release);
heartbeat_msg_t hb = {
.sender_priority = 0,
.term = 6, /* < standby term 7 -> fails term-guard */
.committed_lsn = 99,
.committed_config_tag = 0xBB,
.elected_at_ms = 555555,
};
ha_consume_heartbeat_committed(&c, &hb);
TEST_ASSERT_EQUAL_UINT64(4242, hb_elected(&c)); /* unchanged (3.2 shape) */
hb_teardown_cluster(&c);
printf("test_consume_stale_term_does_not_store_elected_at_ms: PASS\n");
}
/* ── suite entry point ─────────────────────────────────────── */
void test_ha_heartbeat_committed(void)
@ -346,4 +508,8 @@ void test_ha_heartbeat_committed(void)
RUN_TEST(test_consume_stale_term_ignored);
RUN_TEST(test_consume_primary_does_not_consume);
RUN_TEST(test_consume_null_or_zero_noop);
/* controller-status delta (task 7.3): heartbeat-carried election time */
RUN_TEST(test_consume_stores_elected_at_ms);
RUN_TEST(test_consume_elected_at_ms_reconcile_free);
RUN_TEST(test_consume_stale_term_does_not_store_elected_at_ms);
}

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

@ -319,13 +319,14 @@ 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.
* committed_config_tag, elected_at_ms) it stamps onto the heartbeat, so a
* test can assert the heartbeat carries the primary's OWN current committed
* state and election time (Req 12.2). 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);
uint64_t committed_config_tag,
uint64_t elected_at_ms);
/*
* primary_heartbeat_fn_for_test: test entry-point that exercises the primary
@ -359,16 +360,19 @@ 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 uint64_t g_hb_elected_at_ms = 0;
static int stub_heartbeat_req(endpoint_t *ep, uint64_t term,
uint64_t committed_lsn,
uint64_t committed_config_tag)
uint64_t committed_config_tag,
uint64_t elected_at_ms)
{
(void)ep;
g_hb_calls++;
g_hb_term = term;
g_hb_committed_lsn = committed_lsn;
g_hb_committed_config_tag = committed_config_tag;
g_hb_elected_at_ms = elected_at_ms;
return 0;
}
@ -599,11 +603,16 @@ static void test_primary_hb_stamps_own_committed_and_term(void)
const uint64_t PRIMARY_TERM = 7;
const uint64_t PRIMARY_COMMITTED_LSN = 42;
const uint64_t PRIMARY_COMMITTED_TAG = 13;
const uint64_t PRIMARY_ELECTED_AT_MS = 1700000000123ULL;
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);
/* Req 12.2: the primary's own currently-known election time (self-stamped
* by promotion, task 7.2) must be echoed onto the outgoing heartbeat. */
atomic_store_explicit(&c.ha.last_known_elected_at_ms, PRIMARY_ELECTED_AT_MS,
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. */
@ -622,6 +631,7 @@ static void test_primary_hb_stamps_own_committed_and_term(void)
g_hb_calls = 0;
g_hb_term = g_hb_committed_lsn = g_hb_committed_config_tag = 0;
g_hb_elected_at_ms = 0;
ha_heartbeat_req_hook = stub_heartbeat_req;
primary_heartbeat_fn_for_test(&c);
@ -636,6 +646,8 @@ static void test_primary_hb_stamps_own_committed_and_term(void)
"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");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(PRIMARY_ELECTED_AT_MS, g_hb_elected_at_ms,
"Req 12.2: heartbeat must carry the primary's OWN last_known_elected_at_ms");
/* last_push_time still updated on success — bookkeeping unchanged. */
TEST_ASSERT_NOT_EQUAL_MESSAGE(0, (int)sctx->last_push_time,

10
src/controller/testsuite/test_ha_partition.c
View file

@ -44,10 +44,12 @@ extern int (*ha_reconnect_connect_hook)(cluster_t *c, int peer_idx);
extern int primary_heartbeat_fn_for_test(cluster_t *c);
extern int standby_ep_ctx_build_for_test(endpoint_t *ep, ha_state_t *ha,
cluster_t *c, int peer_idx);
/* ha_heartbeat_req_hook: per-standby stamp hook (endpoint, term, committed_lsn, tag). */
/* ha_heartbeat_req_hook: per-standby stamp hook (endpoint, term, committed_lsn,
* tag, elected_at_ms). */
extern int (*ha_heartbeat_req_hook)(endpoint_t *ep, uint64_t term,
uint64_t committed_lsn,
uint64_t committed_config_tag);
uint64_t committed_config_tag,
uint64_t elected_at_ms);
extern int (*ha_nop_req_hook)(endpoint_t *ep);
@ -195,9 +197,11 @@ static int ps_heal_nop_hook(endpoint_t *ep)
* standby_endpoints so it's never reached. Only installed during partition;
* it must never block the tick. */
static int partition_hb_hook(endpoint_t *ep, uint64_t term,
uint64_t committed_lsn, uint64_t committed_config_tag)
uint64_t committed_lsn, uint64_t committed_config_tag,
uint64_t elected_at_ms)
{
(void)ep; (void)term; (void)committed_lsn; (void)committed_config_tag;
(void)elected_at_ms;
atomic_fetch_add(&g_survivor_nops, 1);
return 0;
}