diff --git a/src/controller/ha.c b/src/controller/ha.c index 5d0644bb..928b31a5 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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); } diff --git a/src/controller/ha.h b/src/controller/ha.h index 1abe7680..43bf6fd8 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -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 */ diff --git a/src/controller/testsuite/test_ha_role.c b/src/controller/testsuite/test_ha_role.c index 5c56bd67..c2436c55 100644 --- a/src/controller/testsuite/test_ha_role.c +++ b/src/controller/testsuite/test_ha_role.c @@ -1,6 +1,7 @@ #include #include #include +#include #include @@ -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); }