diff --git a/src/controller/testsuite/test_ha_liveness.c b/src/controller/testsuite/test_ha_liveness.c index 4e3a7fdb..c85f8db0 100644 --- a/src/controller/testsuite/test_ha_liveness.c +++ b/src/controller/testsuite/test_ha_liveness.c @@ -1118,6 +1118,212 @@ static void test_reconnect_offload_dedups_per_peer(void) printf("test_reconnect_offload_dedups_per_peer: PASS\n"); } +/* ── Tests 20–21: raft-drive-async Task 2.3 — reconnect-offload INTEGRATION ── + * + * Acceptance criteria (Req 2.1, 2.2, 2.5; design ReconnectOffload): + * Tests 15–17 prove the SINGLE-tick unit behaviour (one survivor, ≤2 ticks). + * Task 2.3 escalates to SUSTAINED liveness: while ONE peer stays unreachable + * (its reconnect parked in-flight on the lane for the WHOLE window), drive + * MULTIPLE consecutive heartbeat ticks and assert + * - every SURVIVING peer keeps receiving heartbeats on EVERY tick + * (fan-out is observable: a 4-node cluster has TWO survivors), and + * - each tick returns promptly — the delaymgr is NEVER stalled, and + * - the reconnect is deduped across the ENTIRE window (the connect hook is + * submitted exactly ONCE, not once-per-tick — Req 2.5). + * + * RED contrast: reverting the offload (flag OFF = inline connect) makes the + * FIRST tick block in the parked connect, so the tick never returns and + * survivors stop getting heartbeats. The flag toggle is the RED↔GREEN switch + * (production already has the offload; this is a test-only task). + * + * Determinism: same gate as tests 15–17 — the blocking connect is replaced by + * rc_gated_connect_hook, which parks on g_rc_gate, so the test controls exactly + * when the unreachable peer's "connect" completes (no real socket/timing). */ + +#define RC_SUSTAIN_TICKS 3 + +/* Build a 4-node primary (self = priority 0; peers 1,2,3). Peers 1 and 3 are + * reachable standbys already in standby_endpoints; peer 2 is absent, so the + * reconnect scan targets ONLY peer 2 (reconnect_inflight[2]). Two survivors + * make the heartbeat fan-out observable. Modeled on rc_setup_primary (3-node, + * one survivor). */ +static void rc_setup_primary2(cluster_t *c, myconf_t *conf, + endpoint_t *standby_ep1, endpoint_t *standby_ep3) +{ + memset(c, 0, sizeof(*c)); + memset(conf, 0, sizeof(*conf)); + + init_ha_primary_with_standby_eps(&c->ha); + + /* 4-node cluster; self = priority 0. */ + conf->peers_cfg.candidates_count = 4; + for (int i = 0; i < 4; i++) { + snprintf(conf->peers_cfg.candidates[i].name, + sizeof(conf->peers_cfg.candidates[i].name), "ctrl%d", i); + snprintf(conf->peers_cfg.candidates[i].hostname, + sizeof(conf->peers_cfg.candidates[i].hostname), "127.0.0.1"); + conf->peers_cfg.candidates[i].port = 19000 + i; + } + c->myconf = conf; + c->ha.my_priority = 0; + c->myhost = &c->myhost_info; + + /* Slow lane for the offload path. */ + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c->slowlane, 2, 4)); + + /* Reachable standby for peer 1 (absolute index 1), already connected. */ + memset(standby_ep1, 0, sizeof(*standby_ep1)); + standby_ep_ctx_build_for_test(standby_ep1, &c->ha, c, 1); + ((standby_ep_ctx_t_test *)standby_ep1->data)->last_push_time = 0; + zarray_store(c->ha.standby_endpoints, standby_ep1); + + /* Reachable standby for peer 3 (absolute index 3), already connected. + * Peer 2 is intentionally absent → it is the sole reconnect target. */ + memset(standby_ep3, 0, sizeof(*standby_ep3)); + standby_ep_ctx_build_for_test(standby_ep3, &c->ha, c, 3); + ((standby_ep_ctx_t_test *)standby_ep3->data)->last_push_time = 0; + zarray_store(c->ha.standby_endpoints, standby_ep3); +} + +static void rc_teardown_primary2(cluster_t *c, endpoint_t *standby_ep1, + endpoint_t *standby_ep3) +{ + slowlane_fini(&c->slowlane); + free(standby_ep1->data); + free(standby_ep3->data); + fini_ha_primary_with_standby_eps(&c->ha); +} + +/* Drive one heartbeat tick on a worker thread and return 1 iff it completed + * within the deadline (delaymgr not stalled), 0 if it is still blocked. */ +static int rc_drive_tick_bounded(cluster_t *c, pthread_t *th_out, + rc_tick_arg_t *arg) +{ + arg->c = c; + atomic_store(&arg->done, 0); + pthread_create(th_out, NULL, rc_tick_thread, arg); + for (int i = 0; i < 200; i++) { /* up to ~2s */ + if (atomic_load(&arg->done)) return 1; + struct timespec ts = { .tv_sec = 0, .tv_nsec = 10 * 1000000L }; + nanosleep(&ts, NULL); + } + return 0; +} + +/* + * Test 20 / 21 (Req 2.1, 2.2, 2.5): parametrized GREEN/RED runner for the + * sustained reconnect-offload integration behaviour. + */ +static void rc_run_sustained(int offload_on, const char *label) +{ + cluster_t c; + myconf_t conf; + endpoint_t standby_ep1, standby_ep3; + rc_setup_primary2(&c, &conf, &standby_ep1, &standby_ep3); + + rc_gate_init(&g_rc_gate); + ha_reconnect_connect_hook = rc_gated_connect_hook; + ha_set_reconnect_offload_for_test(offload_on); + + g_nop_calls = 0; + ha_nop_req_hook = stub_nop_req; + + standby_ep_ctx_t_test *s1 = (standby_ep_ctx_t_test *)standby_ep1.data; + standby_ep_ctx_t_test *s3 = (standby_ep_ctx_t_test *)standby_ep3.data; + + if (offload_on) { + /* GREEN: drive N consecutive ticks. Reset each survivor's + * last_push_time to 0 BEFORE each tick so a heartbeat is DUE this tick + * (production stamps last_push_time = now after a successful send and + * only re-sends once elapsed >= interval_s; without the reset, ticks 2+ + * would send nothing). */ + for (int t = 0; t < RC_SUSTAIN_TICKS; t++) { + s1->last_push_time = 0; + s3->last_push_time = 0; + + pthread_t th; + rc_tick_arg_t arg; + int done = rc_drive_tick_bounded(&c, &th, &arg); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, done, + "Req 2.1/2.2: every heartbeat tick must return promptly while a " + "peer's reconnect is parked in-flight (delaymgr never stalled)"); + pthread_join(th, NULL); + + /* The unreachable peer's reconnect stays in-flight across the whole + * window (parked in the gate). */ + TEST_ASSERT_EQUAL_INT_MESSAGE(1, + atomic_load_explicit(&c.ha.reconnect_inflight[2], + memory_order_acquire), + "Req 2.5: the unreachable peer's reconnect must remain in-flight " + "(parked) for the whole multi-tick window"); + } + + /* Every surviving peer (2) got a heartbeat on every tick (N) — survivors + * kept getting heartbeats across the WHOLE in-flight window (Req 2.1). */ + TEST_ASSERT_EQUAL_INT_MESSAGE(2 * RC_SUSTAIN_TICKS, g_nop_calls, + "Req 2.1: every surviving peer must keep receiving heartbeats across " + "ALL ticks while one peer is unreachable"); + + /* The offloaded reconnect ran on the lane (hook parked) ... */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc_gate_wait_entered(&g_rc_gate, 1, 1000), + "Req 2.1: the offloaded reconnect connect must run on the lane"); + /* ... exactly ONCE across all ticks — deduped, not once-per-tick. */ + TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_rc_gate.calls), + "Req 2.5: the reconnect to an already-in-flight peer must be deduped " + "across ALL ticks (submitted exactly once, not once per tick)"); + + /* Release the parked reconnect; drain until the in-flight flag clears. */ + rc_gate_open(&g_rc_gate); + struct timespec ts = { .tv_sec = 0, .tv_nsec = 10 * 1000000L }; + for (int i = 0; i < 200; i++) { + if (atomic_load_explicit(&c.ha.reconnect_inflight[2], + memory_order_acquire) == 0) break; + nanosleep(&ts, NULL); + } + } else { + /* RED: offload OFF — the FIRST tick runs the connect INLINE and blocks + * in the gate, so the tick never returns and survivors stop getting + * heartbeats. The flag toggle is what flips RED↔GREEN. */ + s1->last_push_time = 0; + s3->last_push_time = 0; + + pthread_t th; + rc_tick_arg_t arg; + int done = rc_drive_tick_bounded(&c, &th, &arg); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, done, + "RED: with offload OFF the inline connect blocks the delaymgr tick — " + "the tick never returns, so survivors stop getting heartbeats"); + /* The stalled tick cannot sustain heartbeats to survivors across the + * window: it sent at most this single (blocked) tick's fan-out, far + * below the GREEN expectation of 2 survivors × N ticks. */ + TEST_ASSERT_TRUE_MESSAGE(g_nop_calls < 2 * RC_SUSTAIN_TICKS, + "RED: a stalled tick cannot sustain heartbeats to survivors across " + "the multi-tick window"); + + /* Release the parked inline connect so the blocked tick thread finishes, + * then join it. */ + rc_gate_open(&g_rc_gate); + pthread_join(th, NULL); + } + + ha_nop_req_hook = NULL; + ha_reconnect_connect_hook = NULL; + ha_set_reconnect_offload_for_test(1); /* restore production default */ + rc_gate_destroy(&g_rc_gate); + rc_teardown_primary2(&c, &standby_ep1, &standby_ep3); + printf("%s: PASS\n", label); +} + +static void test_reconnect_offload_sustains_survivor_heartbeats(void) +{ + rc_run_sustained(1, "test_reconnect_offload_sustains_survivor_heartbeats"); +} + +static void test_reconnect_inline_stalls_survivor_heartbeats_RED(void) +{ + rc_run_sustained(0, "test_reconnect_inline_stalls_survivor_heartbeats_RED"); +} + /* ── Tests 18–19: raft-drive-async Task 2.2 — inbound/outbound ownership ───── * * Acceptance criteria (Req 2.6, 4.4; design ReconnectOffload race resolution, @@ -1304,5 +1510,8 @@ void test_ha_liveness(void) /* raft-drive-async Task 2.2 — inbound/outbound ownership (Req 2.6, 4.4) */ RUN_TEST(test_reconnect_resolves_dual_channel_same_peer); RUN_TEST(test_reconnect_preserves_other_peer_channel); + /* raft-drive-async Task 2.3 — reconnect-offload INTEGRATION (Req 2.1, 2.2, 2.5) */ + RUN_TEST(test_reconnect_offload_sustains_survivor_heartbeats); + RUN_TEST(test_reconnect_inline_stalls_survivor_heartbeats_RED); UNITY_END(); }