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

fix(ha-repl-hardening): heartbeat re-arm broken by delaymgr ms-task ordering

Browse source 
Root cause: task_cmp in delay.c compared nextrun (always 0 for ms-tasks)
instead of nextrun_ts, making all ms-tasks appear equal in the ordered
queue. The heartbeat task was re-enqueued correctly but stuck behind the
liveness poll task in the broken ordering.

Changes:
- delay.c: fix task_cmp to compare nextrun_ts for ms-tasks (ns precision)
- ha.c: use runcount=-1 (infinite repeat) instead of one-shot+re-arm;
  simpler and avoids the broken re-arm path entirely
- ha.c: add static atomic idempotency guard to prevent duplicate heartbeat
  tasks on re-election
This commit is contained in:
Rongsong Shen 2026-06-06 13:28:00 +00:00
parent f236d320a2
commit e1517bbf28
2 changed files with 34 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

31
src/controller/ha.c
View file

@ -3680,7 +3680,7 @@ static int primary_heartbeat_fn(void *arg)
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
if (role != CTRL_PRIMARY && role != CTRL_PRIMARY_COMMITTING)
return 0; /* demoted — stop the heartbeat loop */
return 0; /* demoted — stop sending NOPs */
/* 1. Heartbeat all connected standbys — idle-triggered NOP (Req 3.1-3.4).
*
@ -3762,12 +3762,8 @@ static int primary_heartbeat_fn(void *arg)
}
}
/* Re-arm for next interval — only reached when still PRIMARY */
int interval_ms = (int)ha->config.heartbeat_interval_ms;
if (interval_ms <= 0)
interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS;
delay_task_ms(&c->delaymgr, interval_ms, 1,
(delay_task_entry_fn_t)primary_heartbeat_fn, c);
/* runcount=-1 (infinite repeat): the delaymgr re-enqueues this task
* automatically. The role check at entry stops work on demotion. */
return 0;
}
@ -3779,17 +3775,30 @@ void ha_start_primary_heartbeat(cluster_t *c)
if (interval_ms <= 0)
interval_ms = HA_DEFAULT_HEARTBEAT_INTERVAL_MS;
/* Idempotency: only arm once per process lifetime. With runcount=-1 the
* delaymgr re-enqueues the task automatically; duplicate tasks would stack
* on every re-election. */
{
static _Atomic int s_armed = 0;
int expected = 0;
if (!atomic_compare_exchange_strong(&s_armed, &expected, 1))
return; /* already armed — task is still running */
}
/* Guard: skip if delaymgr is not running (e.g. unit tests with zeroed cluster_t).
* In production the delaymgr is always started before any promotion. */
if (!c->delaymgr.running)
if (!c->delaymgr.running) {
INFO("HA: heartbeat skipped (delaymgr not running)");
return;
}
/* One-shot: the task re-arms itself each time while role == PRIMARY */
int rc = delay_task_ms(&c->delaymgr, interval_ms, 1,
/* runcount=-1 (infinite): delaymgr re-enqueues automatically.
* The role check in primary_heartbeat_fn no-ops on demotion. */
int rc = delay_task_ms(&c->delaymgr, interval_ms, -1,
(delay_task_entry_fn_t)primary_heartbeat_fn, c);
if (rc < 0)
WARN("ha_start_primary_heartbeat: delay_task_ms failed: %d", rc);
else
INFO("HA: primary heartbeat armed (interval=%dms)", interval_ms);
INFO("HA: primary heartbeat armed (interval=%dms, rc=%d)", interval_ms, rc);
}

20
src/lib/delay.c
View file

@ -99,17 +99,25 @@ static int task_cmp(void* ctx, void* arg1, void* arg2)
(void)ctx;
delay_task_t* t1 = (delay_task_t*)arg1;
delay_task_t* t2 = (delay_task_t*)arg2;
time_t k1 = t1->nextrun;
time_t k2 = t2->nextrun;
if (k1 > k2) {
return 1;
}
/* ms-tasks: compare by nextrun_ts (nanosecond-precision deadline).
* Both delay_task_ms and ordq_put use nextrun==0 as the ms-task marker. */
int t1_is_ms = (t1->nextrun_ts.tv_nsec != 0 || t1->nextrun_ts.tv_sec != 0);
int t2_is_ms = (t2->nextrun_ts.tv_nsec != 0 || t2->nextrun_ts.tv_sec != 0);
if (k1 == k2) {
if (t1_is_ms && t2_is_ms) {
if (t1->nextrun_ts.tv_sec > t2->nextrun_ts.tv_sec) return 1;
if (t1->nextrun_ts.tv_sec < t2->nextrun_ts.tv_sec) return -1;
if (t1->nextrun_ts.tv_nsec > t2->nextrun_ts.tv_nsec) return 1;
if (t1->nextrun_ts.tv_nsec < t2->nextrun_ts.tv_nsec) return -1;
return 0;
}
/* Mixed or seconds-based: fall back to nextrun comparison. */
time_t k1 = t1->nextrun;
time_t k2 = t2->nextrun;
if (k1 > k2) return 1;
if (k1 == k2) return 0;
return -1;
}