diff --git a/.kiro/specs/ha-heartbeat-liveness/tasks.md b/.kiro/specs/ha-heartbeat-liveness/tasks.md index 08acc65..993ea8a 100644 --- a/.kiro/specs/ha-heartbeat-liveness/tasks.md +++ b/.kiro/specs/ha-heartbeat-liveness/tasks.md @@ -29,7 +29,7 @@ - _Requirements: 1.2, 1.4, 1.5, 1.6, 2.1, 2.2, 2.3, 4.2, 4.3, 5.2, 5.3, 5.4, 5.6_ - _Boundary: ha.c (standby liveness path)_ -- [ ] 2.2 (P) Extend primary_heartbeat_fn with per-standby idle tracking +- [x] 2.2 (P) Extend primary_heartbeat_fn with per-standby idle tracking - In `primary_heartbeat_fn`: for each ep in `ha->standby_endpoints`, cast `ep->data` to `standby_ep_ctx_t` and read `sctx->last_push_time` - Compute `elapsed = now - sctx->last_push_time`; if `elapsed >= heartbeat_interval_ms` call `xce_nop_req(ep)` and write `sctx->last_push_time = now`; else skip NOP for this standby - In `on_standby_connected`: write `sctx->last_push_time = clock_gettime(...)` immediately after TLS handshake completes (before ctrlhello send); this counts the ctrlhello as the first push @@ -70,7 +70,7 @@ - _Boundary: ha.c unit tests, startup_validation.c unit tests_ - _Depends: 2.1, 1.2_ -- [ ] 4.2 (P) Unit tests for primary idle NOP suppression +- [x] 4.2 (P) Unit tests for primary idle NOP suppression - Test `primary_heartbeat_fn`: sends NOP to standby when `last_push_time` elapsed ≥ `heartbeat_interval_ms` - Test `primary_heartbeat_fn`: skips NOP when `last_push_time` elapsed < `heartbeat_interval_ms` - Test `on_standby_connected` writes non-zero `last_push_time` immediately diff --git a/src/controller/ha.c b/src/controller/ha.c index a18e94a..4f94f7a 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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 */ diff --git a/src/controller/testsuite/test_ha_liveness.c b/src/controller/testsuite/test_ha_liveness.c index 24cf13c..cd143bd 100644 --- a/src/controller/testsuite/test_ha_liveness.c +++ b/src/controller/testsuite/test_ha_liveness.c @@ -16,12 +16,16 @@ #define _GNU_SOURCE #include #include +#include +#include #include #include #include #include #include "ha.h" +#include "netio.h" +#include "xce/cluster_def.h" /* ── XCE_TESTING seam ────────────────────────────────────────────────────── */ @@ -288,6 +292,198 @@ static void test_ha_disarm_makes_subsequent_poll_noop(void) printf("test_ha_disarm_makes_subsequent_poll_noop: PASS\n"); } +/* ── Task 2.2: primary idle-tracking tests ──────────────────────────────── */ + +/* + * standby_ep_ctx_t: internal ha.c struct; redeclared here for direct + * construction. Must match the layout in ha.c exactly. + */ +typedef struct { + ha_state_t *ha; + cluster_t *c; + void *ctx; /* ha_elect_ctx_t*, always NULL for standby eps */ + int peer_idx; + time_t last_push_time; +} standby_ep_ctx_t_test; /* suffix _test to avoid redefinition; ha.c uses standby_ep_ctx_t */ + +/* + * ha_nop_req_hook: XCE_TESTING hook. When non-NULL, primary_heartbeat_fn + * calls this INSTEAD OF xce_nop_req so tests can count sends without a + * live endpoint. Declared in ha.c. + */ +extern int (*ha_nop_req_hook)(endpoint_t *ep); + +/* + * primary_heartbeat_fn_for_test: test entry-point that exercises the primary + * heartbeat idle-tracking logic directly without a live delaymgr. + * Declared in ha.c under XCE_TESTING. + */ +extern int primary_heartbeat_fn_for_test(cluster_t *c); + +/* + * standby_ep_ctx_build_for_test: allocate a standby_ep_ctx_t, wire ep->data. + * Declared in ha.c under XCE_TESTING. + */ +extern endpoint_t *standby_ep_ctx_build_for_test(endpoint_t *ep, + ha_state_t *ha, + cluster_t *c, + int peer_idx); + +/* ── Helpers ─────────────────────────────────────────────────────────────── */ + +static int g_nop_calls = 0; + +static int stub_nop_req(endpoint_t *ep) +{ + (void)ep; + g_nop_calls++; + return 0; +} + +static void init_ha_primary_with_standby_eps(ha_state_t *ha) +{ + ha_init(ha, &g_cfg, 0); + ha->standby_endpoints = calloc(1, sizeof(zarray_t)); + zarray_init(ha->standby_endpoints, 4, NULL); + /* Set role to PRIMARY so primary_heartbeat_fn does not exit early */ + atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY, memory_order_release); +} + +static void fini_ha_primary_with_standby_eps(ha_state_t *ha) +{ + if (ha->standby_endpoints) { + zarray_fini(ha->standby_endpoints); + free(ha->standby_endpoints); + ha->standby_endpoints = NULL; + } + ha_fini(ha); +} + +/* ── Test 8: primary_heartbeat_fn sends NOP when last_push_time is 0 ────── */ + +/* + * Req 3.1, 3.2: standby ep with last_push_time=0 (long ago) — heartbeat fn + * must send NOP and update last_push_time to a recent timestamp. + */ +static void test_primary_hb_sends_nop_when_idle(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + init_ha_primary_with_standby_eps(&c.ha); + + /* Build a mock standby endpoint with last_push_time=0 (idle since epoch) */ + endpoint_t ep; + memset(&ep, 0, sizeof(ep)); + standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0); + standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data; + sctx->last_push_time = 0; /* long ago — elapsed >> heartbeat_interval_ms */ + + zarray_store(c.ha.standby_endpoints, &ep); + + /* Install NOP hook */ + g_nop_calls = 0; + ha_nop_req_hook = stub_nop_req; + + primary_heartbeat_fn_for_test(&c); + + ha_nop_req_hook = NULL; + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_nop_calls, + "Req 3.2: primary_heartbeat_fn must send NOP when last_push_time is stale"); + + /* last_push_time must have been updated to a recent value */ + TEST_ASSERT_NOT_EQUAL_MESSAGE(0, (int)sctx->last_push_time, + "Req 3.3: last_push_time must be updated after NOP send"); + + free(ep.data); + fini_ha_primary_with_standby_eps(&c.ha); + printf("test_primary_hb_sends_nop_when_idle: PASS\n"); +} + +/* ── Test 9: primary_heartbeat_fn skips NOP when last_push_time is fresh ── */ + +/* + * Req 3.4: standby ep with last_push_time=now — heartbeat fn must NOT send NOP. + */ +static void test_primary_hb_skips_nop_when_fresh(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + init_ha_primary_with_standby_eps(&c.ha); + + endpoint_t ep; + memset(&ep, 0, sizeof(ep)); + standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0); + standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data; + /* Set last_push_time = now so elapsed ≈ 0 < heartbeat_interval_ms/1000 */ + sctx->last_push_time = mono_now(); + + zarray_store(c.ha.standby_endpoints, &ep); + + g_nop_calls = 0; + ha_nop_req_hook = stub_nop_req; + + primary_heartbeat_fn_for_test(&c); + + ha_nop_req_hook = NULL; + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_nop_calls, + "Req 3.4: primary_heartbeat_fn must NOT send NOP when last_push_time is fresh"); + + free(ep.data); + fini_ha_primary_with_standby_eps(&c.ha); + printf("test_primary_hb_skips_nop_when_fresh: PASS\n"); +} + +/* ── Test 10: on_standby_connected sets last_push_time ──────────────────── */ + +/* + * Req 3.3: on_standby_connected must write last_push_time so that the first + * heartbeat tick after connect does not send a redundant NOP. + * + * We use the standby_ep_ctx_build_for_test to construct a context with + * last_push_time=0, then verify that after a fresh connect (simulated by + * calling on_standby_connected_for_test), last_push_time is non-zero. + */ +extern int on_standby_connected_for_test(endpoint_t *ep); + +static void test_on_standby_connected_sets_last_push_time(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + ha_state_t *ha = &c.ha; + ha_init(ha, &g_cfg, 0); + ha->standby_endpoints = calloc(1, sizeof(zarray_t)); + zarray_init(ha->standby_endpoints, 4, NULL); + /* Provide minimal myconf so on_standby_connected doesn't deref NULL */ + myconf_t conf; + memset(&conf, 0, sizeof(conf)); + c.myconf = &conf; + + endpoint_t ep; + memset(&ep, 0, sizeof(ep)); + standby_ep_ctx_build_for_test(&ep, ha, &c, 0); + standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)ep.data; + + /* last_push_time is 0 from calloc */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, (int)sctx->last_push_time, + "last_push_time must start at 0 before on_standby_connected"); + + /* Call on_standby_connected directly via test entry (xce_request will fail + * on a null ep but last_push_time should be written before it) */ + on_standby_connected_for_test(&ep); + + TEST_ASSERT_NOT_EQUAL_MESSAGE(0, (int)sctx->last_push_time, + "Req 3.3: on_standby_connected must write last_push_time != 0"); + + free(ep.data); + zarray_fini(ha->standby_endpoints); + free(ha->standby_endpoints); + ha->standby_endpoints = NULL; + ha_fini(ha); + printf("test_on_standby_connected_sets_last_push_time: PASS\n"); +} + /* ── Suite entry point ──────────────────────────────────────────────────── */ void test_ha_liveness(void) @@ -300,5 +496,8 @@ void test_ha_liveness(void) RUN_TEST(test_liveness_poll_skips_if_not_standby); RUN_TEST(test_ha_arm_no_crash_no_thread); RUN_TEST(test_ha_disarm_makes_subsequent_poll_noop); + RUN_TEST(test_primary_hb_sends_nop_when_idle); + RUN_TEST(test_primary_hb_skips_nop_when_fresh); + RUN_TEST(test_on_standby_connected_sets_last_push_time); UNITY_END(); }