feat: heartbeat watchdog pthread + ha_check_heartbeat + start/stop API (P1-3)

- Add ha_start_watchdog() / ha_stop_watchdog() — dedicated pthread with
  nanosleep(heartbeat_interval_ms) loop; stop via atomic watchdog_stop +
  pthread_join (no pthread_cancel, safe for OpenSSL)
- Add ha_check_heartbeat_nocluster() — reads role with
  atomic_load_explicit(memory_order_acquire), guards on primary_host[0]=='\0',
  computes elapsed_ms from CLOCK_MONOTONIC, sets failover_requested on timeout
- Update ha_fini() to call ha_stop_watchdog() before mutex destroy
- 3 new tests (TDD RED→GREEN): watchdog starts/stops, no-op when no primary,
  stale heartbeat no crash; all 97 controller tests green
This commit is contained in:
Rongsong Shen 2026-05-07 14:19:27 +00:00
parent a289edb777
commit 2fae4c3df5
3 changed files with 165 additions and 1 deletions

View file

@ -37,11 +37,84 @@ int ha_init(ha_state_t *ha, const ha_config_t *cfg, int my_priority)
return 0;
}
/* ── Watchdog thread implementation ────────────────────────── */
void ha_check_heartbeat_nocluster(ha_state_t *ha)
{
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role, memory_order_acquire);
/* Only standbys watch for primary heartbeat timeout */
if (role != CTRL_STANDBY) return;
/* If we have no primary endpoint, nothing to monitor */
if (ha->primary_host[0] == '\0') return;
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
pthread_mutex_lock(&ha->lock);
struct timespec last = ha->last_heartbeat;
pthread_mutex_unlock(&ha->lock);
uint64_t elapsed_ms =
(uint64_t)(now.tv_sec - last.tv_sec) * 1000ULL +
(uint64_t)(now.tv_nsec - last.tv_nsec) / 1000000ULL;
if (elapsed_ms >= ha->config.heartbeat_timeout_ms) {
/* Primary appears dead — full promotion in P1-3a.
* For now: set failover_requested so watchdog thread can act. */
atomic_store_explicit(&ha->failover_requested, 1, memory_order_release);
}
}
static void *ha_watchdog_thread(void *arg)
{
ha_state_t *ha = arg;
while (!atomic_load_explicit(&ha->watchdog_stop, memory_order_acquire)) {
long ms = (long)ha->config.heartbeat_interval_ms;
struct timespec ts = {
.tv_sec = ms / 1000,
.tv_nsec = (ms % 1000) * 1000000L,
};
nanosleep(&ts, NULL);
/* Re-check stop flag after sleep */
if (atomic_load_explicit(&ha->watchdog_stop, memory_order_acquire)) break;
ha_check_heartbeat_nocluster(ha);
/* Handle manual failover request (SIGUSR1 path — P1-3a) */
if (atomic_load_explicit(&ha->failover_requested, memory_order_acquire)) {
atomic_store_explicit(&ha->failover_requested, 0, memory_order_release);
/* Full ha_promote_to_primary(c) call in P1-3a when cluster_t is available */
}
}
return NULL;
}
int ha_start_watchdog(ha_state_t *ha)
{
atomic_store_explicit(&ha->watchdog_stop, 0, memory_order_release);
int rc = pthread_create(&ha->watchdog_thread, NULL, ha_watchdog_thread, ha);
if (rc != 0) return -rc;
return 0;
}
void ha_stop_watchdog(ha_state_t *ha)
{
atomic_store_explicit(&ha->watchdog_stop, 1, memory_order_release);
if (ha->watchdog_thread != 0) {
pthread_join(ha->watchdog_thread, NULL);
ha->watchdog_thread = 0;
}
}
void ha_fini(ha_state_t *ha)
{
if (!ha)
return;
/* Watchdog thread join added in P1-3 */
ha_stop_watchdog(ha); /* join watchdog thread before destroying mutex */
pthread_mutex_destroy(&ha->lock);
}

View file

@ -97,4 +97,25 @@ int ha_promote_to_primary(cluster_t *c);
*/
int ha_demote_to_standby(cluster_t *c);
/* ── Watchdog thread API (P1-3) ────────────────────────────── */
/*
* ha_start_watchdog: spawn the heartbeat watchdog pthread.
* Must be called after ha_init(). Returns 0 on success, -errno on failure.
*/
int ha_start_watchdog(ha_state_t *ha);
/*
* ha_stop_watchdog: set watchdog_stop flag and pthread_join the thread.
* Safe to call if watchdog was never started (thread == 0).
*/
void ha_stop_watchdog(ha_state_t *ha);
/*
* ha_check_heartbeat_nocluster: check whether the primary heartbeat has
* timed out. Sets failover_requested if timeout exceeded and primary_host
* is non-empty. Test-only entry point — does not need cluster_t.
*/
void ha_check_heartbeat_nocluster(ha_state_t *ha);
#endif /* HA_H */

View file

@ -1,6 +1,7 @@
#include <stdatomic.h>
#include <string.h>
#include <stdio.h>
#include <time.h>
#include <unity/unity.h>
@ -84,6 +85,72 @@ static void test_ha_promote_sets_primary_role(void)
printf("test_ha_promote_sets_primary_role: PASS\n");
}
/* ── P1-3 watchdog tests ────────────────────────────────────── */
/* ha_start_watchdog() must be callable after ha_init() */
static void test_watchdog_starts_and_stops(void)
{
ha_state_t ha;
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
};
TEST_ASSERT_EQUAL_INT(0, ha_init(&ha, &cfg, 0));
TEST_ASSERT_EQUAL_INT(0, ha_start_watchdog(&ha));
/* brief sleep to let the thread wake at least once */
struct timespec ts = { 0, 50 * 1000000L }; /* 50ms */
nanosleep(&ts, NULL);
ha_stop_watchdog(&ha); /* sets stop flag + pthread_join */
ha_fini(&ha);
printf("test_watchdog_starts_and_stops: PASS\n");
}
/* ha_check_heartbeat_nocluster(): when in CTRL_STANDBY with no primary_ep,
* must not promote */
static void test_heartbeat_no_action_when_no_primary(void)
{
ha_state_t ha;
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
};
ha_init(&ha, &cfg, 1 /* not highest priority */);
/* No primary_host set — watchdog must not promote */
ha_check_heartbeat_nocluster(&ha); /* test-only accessor */
TEST_ASSERT_EQUAL_INT(CTRL_STANDBY,
(int)atomic_load_explicit(&ha.role, memory_order_acquire));
ha_fini(&ha);
printf("test_heartbeat_no_action_when_no_primary: PASS\n");
}
/* ha_check_heartbeat_nocluster(): when last_heartbeat is stale (timeout exceeded)
* and primary_host is empty (no known primary), must not crash or change role */
static void test_heartbeat_stale_no_crash_without_primary(void)
{
ha_state_t ha;
ha_config_t cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 100, /* very short for test */
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
};
ha_init(&ha, &cfg, 0);
/* Set last_heartbeat to a time far in the past to simulate timeout */
clock_gettime(CLOCK_MONOTONIC, &ha.last_heartbeat);
ha.last_heartbeat.tv_sec -= 10; /* 10 seconds ago */
ha_check_heartbeat_nocluster(&ha);
/* Should not crash. Role may or may not change — but no segfault. */
ha_fini(&ha);
printf("test_heartbeat_stale_no_crash_without_primary: PASS\n");
}
void test_ha_role(void)
{
RUN_TEST(test_ha_init_default_role_is_standby);
@ -91,4 +158,7 @@ void test_ha_role(void)
RUN_TEST(test_ha_init_rejects_short_election_timeout);
RUN_TEST(test_ha_init_rejects_timeout_less_than_interval);
RUN_TEST(test_ha_promote_sets_primary_role);
RUN_TEST(test_watchdog_starts_and_stops);
RUN_TEST(test_heartbeat_no_action_when_no_primary);
RUN_TEST(test_heartbeat_stale_no_crash_without_primary);
}