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

feat(ha-single-leader-role): direct CTRL_PRIMARY promotion + own-term gate via shared sequence (task 2.2)

Browse source
This commit is contained in:
Rongsong Shen 2026-06-13 09:26:08 +00:00
parent b7a69afa0b
commit c089fb69f3
3 changed files with 177 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

171
src/controller/ha.c
View file

@ -90,6 +90,10 @@ extern int push_config_transport_only(cluster_t *c, const uint8_t *bytes,
static void on_vote_connected(endpoint_t *ep);
static int vote_ep_err_handler(void *ep_raw);
static int committing_timeout_fn(void *arg);
/* ha-single-leader-role task 2.2: shared promotion seam — every road to
* PRIMARY registers the persist hooks and arms the own-term gate. */
static void ha_register_persist_hooks(cluster_t *c);
static void ha_arm_own_term_gate(cluster_t *c);
static int ha_heartbeat_is_fresh(ha_state_t *ha);
static int primary_heartbeat_fn(void *arg);
static void on_standby_connected(endpoint_t *ep);
@ -510,15 +514,19 @@ static void raft_cb_become_leader(void *ctx, uint64_t term)
ha->phase2_done = 1;
}
/* Enter COMMITTING: block job traffic until majority ack */
atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY_COMMITTING,
/* ha-single-leader-role (Req 1.1): enter CTRL_PRIMARY directly — there is
* no transitional COMMITTING state. New-leader safety comes from the
* own-term commit rule (the WAL_NOOP + writable_lsn gate armed below),
* not from a fourth role value. The raft-owned term was already synced
* at entry (~line 463), so we do NOT call ha_do_promote here (it would
* double-increment the term, research.md §4 finding 2). This store sits
* OUTSIDE the c != NULL guard so the c==NULL test shim
* (raft_cb_become_leader_for_test) reaches it. */
atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY,
memory_order_release);
INFO("HA FSM (raft): → CTRL_PRIMARY_COMMITTING (term=%" PRIu64
INFO("HA FSM (raft): → CTRL_PRIMARY (term=%" PRIu64
", first_boot=%d)", term, is_first_boot);
/* Reset ack counter for this committing round */
atomic_store(&ha->repl_ack_count, 0);
/* Req 2.1 (ha-repl-hardening): re-initialize per-standby convergence
* tracking for THIS committing round. Stale converged flags from a
* previous term (demote → re-elect cycle) must not satisfy the
@ -593,37 +601,23 @@ static void raft_cb_become_leader(void *ctx, uint64_t term)
#else
ha_send_ctrlhello_to_all_standbys(c);
#endif
/* Start primary heartbeat immediately in COMMITTING so standbys receive
* NOP during the committing window and their liveness poll does not fire.
* ha_start_primary_heartbeat is also called in ha_promote_to_primary;
* the idempotency guard (role check at fn top) prevents double-arming. */
/* ha-single-leader-role task 2.2: register the persistence hooks FIRST,
* before the own-term no-op append, so the persistence seam is live
* before any racing client write can reach the cluster. Idempotent
* via the hooks_registered guard. */
ha_register_persist_hooks(c);
/* Start primary heartbeat immediately so standbys receive NOP and their
* liveness poll does not fire. ha_start_primary_heartbeat is also
* called in ha_promote_to_primary; the idempotency guard (role check
* at fn top) prevents double-arming. */
ha_start_primary_heartbeat(c);
/* Arm committing timeout watchdog */
{
/* Cancel any prior committing timeout (e.g. from a previous term) */
committing_timeout_ctx_t *prev = (committing_timeout_ctx_t *)
atomic_exchange(&ha->committing_timeout_ctx, NULL);
if (prev) {
atomic_store(&prev->cancelled, 1);
}
}
committing_timeout_ctx_t *ct = calloc(1, sizeof(*ct));
if (ct) {
atomic_init(&ct->cancelled, 0);
ct->ha = ha;
ct->c = c;
atomic_store(&ha->committing_timeout_ctx, ct);
int rc = delay_task_ms(&c->delaymgr,
(int)ha->config.committing_timeout_ms,
1, committing_timeout_fn, ct);
if (rc < 0) {
WARN("raft_cb_become_leader: delay_task_ms for committing timeout failed: %d",
rc);
atomic_store(&ha->committing_timeout_ctx, NULL);
free(ct);
}
}
/* ha-single-leader-role task 2.2 (Req 1.4, 2.1, 2.2): append the
* own-term WAL_NOOP and arm the commit engine's writable_lsn gate.
* This is the shared promotion tail both roads to PRIMARY run.
* Non-blocking: promotion never waits for the no-op to commit. */
ha_arm_own_term_gate(c);
}
}
@ -1709,30 +1703,107 @@ int ha_no_primary_known(const ha_state_t *ha)
}
/*
* ha_promote_to_primary: increment term, set role=PRIMARY, then register
* persistence hooks (idempotency guard via hooks_registered).
* Safe to call on a zero-init cluster_t (zeroed hooksys → xce_add_hook
* returns -EINVAL harmlessly; hooks_registered is still set to 1).
* ha_register_persist_hooks: register the three persistence hooks on the
* cluster's hook system, guarded by the hooks_registered idempotency flag.
*
* ha-single-leader-role task 2.2: extracted from ha_promote_to_primary so
* BOTH roads to PRIMARY (the raft become-leader callback and the non-raft
* ha_promote_to_primary path) share one hook-registration sequence and
* cannot diverge. Safe on a zero-init cluster_t (zeroed hooksys →
* xce_add_hook returns -EINVAL harmlessly; hooks_registered still set to 1).
* Requires non-NULL c (dereferences &c->hooksys).
*/
static void ha_register_persist_hooks(cluster_t *c)
{
if (c->ha.hooks_registered)
return;
xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_JOB,
(hook_fn_t)persist_job_hook_fun, &c->persist);
xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_QUEUE,
(hook_fn_t)persist_queue_hook_fun, &c->persist);
xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER,
(hook_fn_t)persist_worker_hook_fun, &c->persist);
c->ha.hooks_registered = 1;
}
/*
* ha_arm_own_term_gate: append the own-term WAL_NOOP and arm the commit
* engine's writable_lsn gate (ha-single-leader-role task 2.2, Req 1.4/2.1/2.2).
*
* The shared promotion tail both roads to PRIMARY run. Sequence:
* 1. persist_wal_async(WAL_NOOP) — appends a header-only entry and
* broadcasts it to standbys via repl_broadcast (one call does both).
* 2. derive noop_lsn = current_lsn - 1 under p->lock (the append
* post-incremented current_lsn; persist.c:1019-1021 pattern).
* 3. commit_submit(noop_lsn, COMMIT_MODE_RETAIN) — lock-free, non-blocking,
* registration-free observability bookkeeping (records nothing).
* 4. commit_set_writable_lsn(noop_lsn) — arms the own-term gate so the
* engine withholds WAIT completion until committed_lsn >= noop_lsn.
*
* Non-blocking by construction: every call here returns without waiting for
* the no-op to commit (Req 1.4 — status honesty does not wait).
*
* Degraded-but-safe (design §Error Strategy): if the no-op append fails,
* no LSN exists, so the gate stays UNSET; promotion STILL COMPLETES. The
* universal commit barrier still requires per-entry majority of a new-term
* entry, so the first state-changing entry of this term becomes the de-facto
* own-term vehicle. We do NOT block and do NOT abort promotion.
*/
static void ha_arm_own_term_gate(cluster_t *c)
{
int rc = persist_wal_async(&c->persist, WAL_NOOP, NULL, 0);
if (rc < 0) {
ERROR("HA: own-term WAL_NOOP append failed: %d", rc);
WARN("HA: own-term gate left UNSET (degraded-but-safe); first "
"new-term entry becomes the own-term commit vehicle");
return;
}
/* Derive the no-op's LSN: the append post-incremented current_lsn, so the
* entry just written took current_lsn - 1 (read under p->lock). */
uint64_t noop_lsn;
pthread_mutex_lock(&c->persist.lock);
noop_lsn = c->persist.current_lsn - 1;
pthread_mutex_unlock(&c->persist.lock);
/* Observability-only bookkeeping (lock-free, non-blocking, records
* nothing per entry — commit state derives from registry watermarks). */
(void)commit_submit(&c->commit, noop_lsn, COMMIT_MODE_RETAIN, 0);
/* Arm the own-term gate: completion of any WAIT is withheld until
* committed_lsn >= noop_lsn. */
commit_set_writable_lsn(&c->commit, noop_lsn);
INFO("HA: own-term noop appended lsn=%" PRIu64 ", writable_lsn gate armed",
noop_lsn);
}
/*
* ha_promote_to_primary: the NON-RAFT promotion road (cluster.c init-time,
* legacy XCE_TESTING evaluation path, tests). This path OWNS its term, so it
* calls ha_do_promote (which increments the term + sets role=PRIMARY). It
* then runs the SAME shared sequence the raft road runs: register hooks,
* start the heartbeat, and arm the own-term gate (noop + writable_lsn).
*
* Every road to PRIMARY arms the own-term gate. The noop+gate tail is
* guarded on c != NULL (callers always pass a real cluster here).
*/
int ha_promote_to_primary(cluster_t *c)
{
ha_do_promote(&c->ha);
ha_do_promote(&c->ha); /* non-raft path owns its term increment */
/* Register persistence hooks — idempotency guard prevents double-add */
if (c->ha.hooks_registered == 0) {
xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_JOB,
(hook_fn_t)persist_job_hook_fun, &c->persist);
xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_QUEUE,
(hook_fn_t)persist_queue_hook_fun, &c->persist);
xce_add_hook(&c->hooksys, XCE_CONTROLLER_HOOK_WORKER,
(hook_fn_t)persist_worker_hook_fun, &c->persist);
c->ha.hooks_registered = 1;
}
/* Shared promotion sequence — must match the raft become-leader tail:
* hooks live BEFORE the no-op append so the persistence seam is active
* before any racing client write. */
ha_register_persist_hooks(c);
/* Arm periodic heartbeat to standbys — keeps their watchdog alive when
* the cluster is idle (no WAL entries or config pushes). */
ha_start_primary_heartbeat(c);
/* Own-term no-op + writable_lsn gate (Req 1.4, 2.1, 2.2). */
ha_arm_own_term_gate(c);
return 0;
}

55
src/controller/testsuite/test_ha_role_lifecycle.c
View file

@ -59,6 +59,27 @@
* replaces these with the WAL_NOOP symbol once it lands. */
#define WAL_NOOP_VALUE 40u
/* ------------------------------------------------------------------ */
/* Broadcast drop-hook (precedent: test_p2_integration.c:32/313, */
/* test_commit_engine.c:1133/1308; hook defined in repl.c:21). */
/* The lifecycle fixture registers fake int g_lc_eps[] endpoints via */
/* repl_add_standby; the own-term WAL_NOOP append now broadcasts via */
/* repl_broadcast → xce_wal_batch_req_raw(ep), which would dereference */
/* ep->peer past the 4-byte int (ASAN global-buffer-overflow at */
/* request.c). Installing this hook drops the wire send harmlessly. */
/* ------------------------------------------------------------------ */
#ifdef XCE_TESTING
extern int (*repl_broadcast_send_hook)(void *ep,
wal_catchup_entry_t **entries,
int n_entries);
static int lc_drop_send_hook(void *ep, wal_catchup_entry_t **entries,
int n_entries)
{
(void)ep; (void)entries; (void)n_entries;
return 0; /* pretend success, drop the data */
}
#endif
/* ------------------------------------------------------------------ */
/* Test-local fixtures */
/* ------------------------------------------------------------------ */
@ -652,6 +673,12 @@ static uint64_t lifecycle_cluster_setup(const char *suffix, int n_peers,
commit_engine_init(&g_lc.commit, &g_lc.persist, n_peers, n_vcpus));
repl_set_commit_engine(&g_lc.persist.repl, &g_lc.commit);
/* Drop wire sends from the own-term WAL_NOOP broadcast — the fake int
* endpoints have no real ep->peer (see lc_drop_send_hook above). */
#ifdef XCE_TESTING
repl_broadcast_send_hook = lc_drop_send_hook;
#endif
/* delaymgr initialised but deliberately NOT started: promotion's heartbeat
* arm is skipped (ha_start_primary_heartbeat guards on delaymgr.running),
* so any task that appears in tasks.head was armed by promotion itself. */
@ -662,6 +689,9 @@ static uint64_t lifecycle_cluster_setup(const char *suffix, int n_peers,
static void lifecycle_cluster_teardown(void)
{
#ifdef XCE_TESTING
repl_broadcast_send_hook = NULL;
#endif
commit_engine_fini(&g_lc.commit);
delaymgr_fini(&g_lc.delaymgr);
persist_stop(&g_lc.persist);
@ -841,10 +871,16 @@ static void test_promotion_appends_noop_and_arms_own_term_gate(void)
* N=3 (current_lsn-1). Entries 1..3 are appended into the engine so a
* waiter on entry M=1 has a target below the gate. */
const uint64_t noop_lsn = 3;
/* persist_wal_async post-increments current_lsn (persist.c:318), and
* production derives noop_lsn = current_lsn - 1. The fixture primes
* current_lsn = appended_lsn + 1, so passing appended_lsn = noop_lsn - 1
* lands the no-op at exactly lsn = noop_lsn. */
(void)lifecycle_cluster_setup("gate", /*n_peers=*/3, /*n_standbys=*/2,
/*appended_lsn=*/noop_lsn, /*n_vcpus=*/8);
/* Register appended entries 1..3 (RETAIN = lock-free bookkeeping). */
for (uint64_t lsn = 1; lsn <= noop_lsn; lsn++)
/*appended_lsn=*/noop_lsn - 1, /*n_vcpus=*/8);
/* Register pre-existing entries strictly BELOW the gate (1..noop_lsn-1);
* the no-op itself occupies noop_lsn. A waiter on M=1 still has a
* target below the gate. */
for (uint64_t lsn = 1; lsn < noop_lsn; lsn++)
(void)commit_submit(&g_lc.commit, lsn, COMMIT_MODE_RETAIN, 0);
/* Act: drive the shared promotion sequence (appends WAL_NOOP + gate). */
@ -921,9 +957,11 @@ static void test_pre_own_term_wait_has_no_third_path(void)
{
#ifdef XCE_TESTING
const uint64_t noop_lsn = 3;
/* appended_lsn = noop_lsn - 1 lands the no-op at lsn = noop_lsn (see
* case 5 comment); entries 1..noop_lsn-1 sit strictly below the gate. */
(void)lifecycle_cluster_setup("nothird", /*n_peers=*/3, /*n_standbys=*/2,
/*appended_lsn=*/noop_lsn, /*n_vcpus=*/8);
for (uint64_t lsn = 1; lsn <= noop_lsn; lsn++)
/*appended_lsn=*/noop_lsn - 1, /*n_vcpus=*/8);
for (uint64_t lsn = 1; lsn < noop_lsn; lsn++)
(void)commit_submit(&g_lc.commit, lsn, COMMIT_MODE_RETAIN, 0);
/* Act: promotion (must arm gate = noop_lsn). */
@ -982,10 +1020,11 @@ static void test_standalone_promotion_gate_opens_immediately(void)
{
#ifdef XCE_TESTING
const uint64_t noop_lsn = 1; /* standalone: noop is the first entry */
/* appended_lsn = 0 → current_lsn primed to 1 → the no-op lands at lsn=1
* (= noop_lsn). No entries exist below the no-op, so there is no
* pre-existing RETAIN submit. */
(void)lifecycle_cluster_setup("solo", /*n_peers=*/1, /*n_standbys=*/0,
/*appended_lsn=*/noop_lsn, /*n_vcpus=*/8);
/* Register the appended entry the no-op occupies. */
(void)commit_submit(&g_lc.commit, noop_lsn, COMMIT_MODE_RETAIN, 0);
/*appended_lsn=*/0, /*n_vcpus=*/8);
/* Act */
TEST_ASSERT_EQUAL_INT(0, ha_promote_to_primary(&g_lc));