wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

feat(ha-heartbeat-liveness): idle-triggered primary NOP per standby (task 2.2)

Browse source 
Extend primary_heartbeat_fn to read last_push_time per standby and send NOP
only when elapsed >= heartbeat_interval_s. on_standby_connected writes
last_push_time before ctrlhello send. Remove workaround xce_nop_req at end
of on_standby_connected. Add ha_nop_req_hook XCE_TESTING seam. 3 new tests.
This commit is contained in:
Rongsong Shen 2026-06-01 15:05:41 +00:00
parent 738859fb4e
commit 622302a55c
3 changed files with 298 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

113
src/controller/ha.c
View file

@ -82,6 +82,8 @@ static void on_vote_connected(endpoint_t *ep);
static int vote_ep_err_handler(void *ep_raw);
static int committing_timeout_fn(void *arg);
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);
/* ── Real raft callback implementations (Task 6.2) ─────────────
*
@ -1566,6 +1568,13 @@ int (*ha_send_ctrlhello_hook)(void *c) = NULL;
*/
int (*ha_push_config_hook)(void *c) = NULL;
/*
* ha_nop_req_hook: test-only seam for xce_nop_req in primary_heartbeat_fn.
* When non-NULL, primary_heartbeat_fn calls this INSTEAD OF xce_nop_req so
* tests can count NOP sends without a live endpoint. Req 3.2, task 2.2.
*/
int (*ha_nop_req_hook)(endpoint_t *ep) = NULL;
#endif /* XCE_TESTING */
/* ── Election Evaluation (Req 2.1, 2.7, 2.8) ───────────────────────────────
@ -2830,6 +2839,16 @@ void liveness_poll_fn_test_entry(liveness_poll_ctx_t *ctx)
{
liveness_poll_fn(ctx);
}
/*
* primary_heartbeat_fn_for_test: call primary_heartbeat_fn directly with c.
* Bypasses delaymgr scheduling so tests can drive the idle-NOP logic without
* a live timer. Only compiled when XCE_TESTING is defined. (Task 2.2)
*/
int primary_heartbeat_fn_for_test(cluster_t *c)
{
return primary_heartbeat_fn(c);
}
#endif
void ha_arm_heartbeat_timeout_vtask(ha_state_t *ha, cluster_t *c)
@ -2915,6 +2934,40 @@ typedef struct {
time_t last_push_time;
} standby_ep_ctx_t;
#ifdef XCE_TESTING
/*
* standby_ep_ctx_build_for_test: allocate a standby_ep_ctx_t, wire ep->data,
* and return ep. last_push_time is initialised to 0 (calloc).
* Caller must free(ep->data) after the test. (Task 2.2)
*/
endpoint_t *standby_ep_ctx_build_for_test(endpoint_t *ep,
ha_state_t *ha,
cluster_t *c,
int peer_idx)
{
standby_ep_ctx_t *sctx = calloc(1, sizeof(*sctx));
if (!sctx) return NULL;
sctx->ha = ha;
sctx->c = c;
sctx->ctx = NULL;
sctx->peer_idx = peer_idx;
/* last_push_time = 0 from calloc — tests set it explicitly */
ep->data = sctx;
return ep;
}
/*
* on_standby_connected_for_test: call on_standby_connected directly.
* ep->data must be a valid standby_ep_ctx_t (use standby_ep_ctx_build_for_test).
* xce_request will fail on a null fd — tests only check last_push_time. (Task 2.2)
*/
int on_standby_connected_for_test(endpoint_t *ep)
{
on_standby_connected(ep);
return 0;
}
#endif /* XCE_TESTING */
/*
* on_standby_connected: iomgr on_connected callback for outbound standby
* endpoints created by the heartbeat task.
@ -2941,6 +2994,14 @@ static void on_standby_connected(endpoint_t *ep)
pthread_mutex_unlock(&ha->lock);
}
/* Record last_push_time before any send so the first heartbeat tick after
* connect does not immediately send a redundant NOP (Req 3.3, task 2.2). */
{
struct timespec _ts;
clock_gettime(CLOCK_MONOTONIC, &_ts);
sctx->last_push_time = _ts.tv_sec;
}
/* Send ctrlhello — triggers handle_ctrlhello on the standby side */
uint64_t lsn = atomic_load_explicit(&c->persist.committed_lsn,
memory_order_acquire);
@ -2962,11 +3023,6 @@ static void on_standby_connected(endpoint_t *ep)
INFO("HA: outbound standby connection to peer[%d] established, ctrlhello sent",
sctx->peer_idx);
/* Send an immediate NOP so the standby's heartbeat watchdog deadline resets
* now rather than waiting up to heartbeat_interval_ms for the first periodic
* heartbeat — avoids a false timeout when committing_timeout_ms ≈ heartbeat_timeout_ms. */
xce_nop_req(ep);
/* Set state-aware dispatcher for subsequent messages on this endpoint */
ep->msg_handler = peer_ctrl_msg_handler;
}
@ -3072,19 +3128,44 @@ static int primary_heartbeat_fn(void *arg)
if (role != CTRL_PRIMARY && role != CTRL_PRIMARY_COMMITTING)
return 0; /* demoted — stop the heartbeat loop */
/* 1. Heartbeat all connected standbys */
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
int n = ha->standby_endpoints->count;
for (int i = 0; i < n; i++) {
endpoint_t *ep = (endpoint_t *)zarray_get(ha->standby_endpoints, i);
if (ep) {
int rc = xce_nop_req(ep);
if (rc < 0)
WARN("HA: heartbeat nop to standby[%d] failed: %d", i, rc);
/* 1. Heartbeat all connected standbys — idle-triggered NOP (Req 3.1-3.4).
*
* For each standby ep, compute elapsed time since the last outbound push.
* Send NOP only when elapsed >= heartbeat_interval_ms so that WAL entries,
* config pushes, and ctrlhello suppress redundant NOPs (Req 3.4).
*/
{
struct timespec _hb_ts;
clock_gettime(CLOCK_MONOTONIC, &_hb_ts);
time_t now = _hb_ts.tv_sec;
int interval_s = (int)(ha->config.heartbeat_interval_ms / 1000);
if (interval_s <= 0)
interval_s = HA_DEFAULT_HEARTBEAT_INTERVAL_MS / 1000;
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
int n = ha->standby_endpoints->count;
for (int i = 0; i < n; i++) {
endpoint_t *ep = (endpoint_t *)zarray_get(ha->standby_endpoints, i);
if (!ep) continue;
standby_ep_ctx_t *sctx = (standby_ep_ctx_t *)ep->data;
if (!sctx) continue;
time_t elapsed = now - sctx->last_push_time;
if (elapsed >= (time_t)interval_s) {
#ifdef XCE_TESTING
int rc = ha_nop_req_hook ? ha_nop_req_hook(ep) : xce_nop_req(ep);
#else
int rc = xce_nop_req(ep);
#endif
if (rc < 0)
WARN("HA: heartbeat nop to standby[%d] failed: %d", i, rc);
else
sctx->last_push_time = now;
}
}
pthread_mutex_unlock(&ha->lock);
}
pthread_mutex_unlock(&ha->lock);
}
/* 2. Reconnect to any peer missing from standby_endpoints */