wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

test(raft-drive-async): sustained reconnect-offload integration test (task 2.3)

Browse source 
Escalates the single-tick unit tests (15-17) to sustained liveness: with
one peer unreachable (reconnect parked in-flight on the lane for the whole
window), drives 3 consecutive heartbeat ticks against a 4-node cluster
(two survivors) and asserts every survivor keeps receiving heartbeats on
every tick (g_nop_calls == 2*3), each tick returns promptly (delaymgr
never stalled), the reconnect stays in-flight throughout, and is deduped
across all ticks (connect hook submitted exactly once — Req 2.5).

RED arm (offload flag OFF) proves the inline connect blocks the first
tick, so survivors stop getting heartbeats (g_nop_calls < 6). The flag
toggle is the RED<->GREEN switch; production is unchanged (test-only task).

test_ha_liveness suite 23 -> 25 tests; full controller suite 196/196 pass.
This commit is contained in:
Rongsong Shen 2026-06-26 10:04:57 +00:00
parent e03ab9bd72
commit 21e26a17b6
1 changed files with 209 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

209
src/controller/testsuite/test_ha_liveness.c
View file

@ -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();
}