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

feat(raft-drive-async): resolve inbound/outbound channel ownership on reconnect (task 2.2)

Browse source 
When a reconnect registers a new outbound control channel for peer X,
on_standby_connected now evicts any pre-existing standby_endpoints entry
for the same peer_idx (a promoted vote-ep or a stale prior channel),
keeping exactly one live outbound channel per peer (Req 2.6/4.4). The
stale ep is torn down OUTSIDE ha->lock via standby_ep_err_handler_primary
(which re-takes the lock) and reclaimed; resolution is strictly per-peer.

Also fixes endpoint leaks on the endpoint_get() calloc paths: every
endpoint_fini() error path now free(ep) (fini reclaims inner members
only), closing the task-2.1 GREEN-test leak masked by detect_leaks=0.

Tests 18-19 in test_ha_liveness.c assert single-survivor resolution and
per-peer isolation. 196/196 controller unit tests pass.
This commit is contained in:
Rongsong Shen 2026-06-26 09:54:10 +00:00
parent e07667ff9b
commit e03ab9bd72
2 changed files with 259 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

102
src/controller/ha.c
View file

@ -108,6 +108,7 @@ static void ha_arm_own_term_gate(cluster_t *c);
int ha_heartbeat_is_fresh(ha_state_t *ha); /* exposed for PreVote denial-reason logging (cluster.c) */
static int primary_heartbeat_fn(void *arg);
static void on_standby_connected(endpoint_t *ep);
static int standby_ep_err_handler_primary(void *ep_raw);
/* ── Real raft callback implementations (Task 6.2) ─────────────
*
@ -3853,6 +3854,16 @@ int (*ha_heartbeat_req_hook)(endpoint_t *ep, uint64_t term,
*/
int (*ha_reconnect_connect_hook)(cluster_t *c, int peer_idx) = NULL;
#ifdef XCE_TESTING
/*
* ha_standby_evicted_count_for_test: counts how many stale same-peer outbound
* channels on_standby_connected evicted while resolving the inbound-vs-outbound
* connection-ownership race (raft-drive-async Task 2.2, Req 2.6). Test-only
* observability — never read on the production path.
*/
_Atomic int ha_standby_evicted_count_for_test = 0;
#endif
/*
* ha_repl_push_config_for_standby_hook: test-only seam for the per-standby
* config push triggered by the heartbeat-reply decision loop (Task 3.3).
@ -6318,6 +6329,17 @@ int on_standby_connected_for_test(endpoint_t *ep)
on_standby_connected(ep);
return 0;
}
/*
* standby_ep_err_handler_primary_for_test: expose the outbound-standby error
* handler so a test can tear down a heap endpoint the production way (remove
* from standby_endpoints + replicator, free ep->data) before reclaiming the
* endpoint itself. (raft-drive-async Task 2.2)
*/
int standby_ep_err_handler_primary_for_test(endpoint_t *ep)
{
return standby_ep_err_handler_primary(ep);
}
#endif /* XCE_TESTING */
/*
@ -6334,18 +6356,76 @@ static void on_standby_connected(endpoint_t *ep)
ha_state_t *ha = sctx->ha;
cluster_t *c = sctx->c;
/* Register in standby_endpoints under lock (idempotency guard) */
/*
* Register in standby_endpoints, resolving the inbound-vs-outbound
* connection-ownership race (raft-drive-async Task 2.2, Req 2.6/4.4;
* ha-async-io.md connection ownership).
*
* The hazard: BOTH raft_cb_become_leader (promotes surviving vote eps) and
* this path (late-joiner / reconnect) zarray_store into standby_endpoints
* keyed only by ep-POINTER identity. So while an outbound reconnect to
* peer X is in flight, a DIFFERENT ep for the same peer X (a just-promoted
* vote-ep, or a stale prior channel whose RDHUP has not yet been processed)
* may already be registered. Leaving both would create two live outbound
* state-push channels to one peer — heartbeats/WAL fan out to a dead/stale
* duplicate (lost/reordered messages, Req 2.6) and the older endpoint is
* orphaned in the replicator.
*
* Resolution (deterministic, per-peer): the freshly-connected channel wins.
* Under ha->lock collect any DIFFERENT ep with the same peer_idx and drop
* it from the array; then register `ep`. The actual teardown of the stale
* ep runs AFTER releasing ha->lock because standby_ep_err_handler_primary
* re-takes ha->lock (and touches the replicator) — avoiding self-deadlock
* and holding the lock across no I/O (R4.4: iomgr/registration state mutated
* only via the lock-guarded interface, never racing the event loop).
*/
endpoint_t *stale_same_peer = NULL;
if (ha->standby_endpoints) {
pthread_mutex_lock(&ha->lock);
int already = 0;
for (int i = 0; i < ha->standby_endpoints->count; i++) {
if (zarray_get(ha->standby_endpoints, i) == ep) { already = 1; break; }
endpoint_t *cur = (endpoint_t *)zarray_get(ha->standby_endpoints, i);
if (cur == ep) { already = 1; continue; }
/* Different ep, same peer → stale duplicate to evict. */
standby_ep_ctx_t *cur_sctx =
cur ? (standby_ep_ctx_t *)cur->data : NULL;
if (cur_sctx && cur_sctx->peer_idx == sctx->peer_idx) {
stale_same_peer = cur;
zarray_del(ha->standby_endpoints, i);
/* count shrank by one; the surviving scan continues but we
* only ever expect a single same-peer duplicate. */
break;
}
}
if (!already)
zarray_store(ha->standby_endpoints, ep);
pthread_mutex_unlock(&ha->lock);
}
/* Tear down the evicted stale channel OUTSIDE ha->lock. Mirror the library
* disconnect path (handle_error → iomgr_unmonitor → err_handler) so we
* never free an endpoint whose fd is still in the epoll set (R4.4: mutate
* iomgr/registration state ONLY via the lock-guarded interface, never
* racing the event loop). err-handler removes it from standby_endpoints +
* the replicator and frees ep->data; we then reclaim the endpoint itself.
* Never self-terminate — this is routine ownership reconciliation. */
if (stale_same_peer) {
WARN("HA: peer[%d] reconnected — evicting stale duplicate outbound "
"channel to keep exactly one live connection (Req 2.6)",
sctx->peer_idx);
if (stale_same_peer->iomgr)
iomgr_unmonitor((iomgr_t *)stale_same_peer->iomgr, stale_same_peer);
(void)standby_ep_err_handler_primary(stale_same_peer);
/* err-handler set ep->data = NULL and freed sctx; endpoint_fini then
* reclaims outq/ssl/packets, and we free the calloc'd endpoint itself
* (endpoint_get uses calloc; there is no endpoint_put). */
endpoint_fini(stale_same_peer);
free(stale_same_peer);
#ifdef XCE_TESTING
atomic_fetch_add(&ha_standby_evicted_count_for_test, 1);
#endif
}
/* Reset failure counter so the next disconnect restarts logging from attempt 1. */
ha->peer_connect_fail_count[sctx->peer_idx] = 0;
@ -6503,6 +6583,7 @@ static reconnect_ctx_t *ha_connect_prep_peer(cluster_t *c, int peer_idx)
if (!rc_ctx) {
ep->data = NULL;
endpoint_fini(ep);
free(ep); /* endpoint_get() calloc'd it; fini frees inner only */
free(sctx);
return NULL;
}
@ -6549,11 +6630,17 @@ static int ha_connect_finish_peer(reconnect_ctx_t *rc_ctx)
* or return a synthetic success/failure without a live socket. On
* the test path we do NOT touch the real endpoint fd. */
int hrc = ha_reconnect_connect_hook(c, peer_idx);
if (hrc < 0) {
/* The hook stands in for the real connect+iomgr_monitor: it NEVER
* registers the synthetic endpoint with iomgr, so the prepped ep/sctx
* are unowned on BOTH success and failure here — reclaim them in either
* case (test-only path; production reclaims via iomgr/err-handler).
* Reverting this to free only on failure re-introduces the 2.1
* GREEN-test endpoint leak masked by detect_leaks=0. */
ep->data = NULL;
endpoint_fini(ep);
free(ep); /* endpoint_get() calloc'd ep; fini frees its inner
* members only, not the struct itself */
free(sctx);
}
return hrc;
}
#endif
@ -6570,9 +6657,11 @@ static int ha_connect_finish_peer(reconnect_ctx_t *rc_ctx)
WARN("HA: connect to peer[%d] (%s:%u) failed: %d",
peer_idx, peer->hostname, peer->port, rc);
/* endpoint_fini calls FREEUP(ep->data) which frees sctx — do not
* double-free sctx here. */
* double-free sctx here. endpoint_get() calloc'd ep itself, so free
* the struct after fini (fini reclaims only its inner members). */
ep->data = NULL;
endpoint_fini(ep);
free(ep);
free(sctx);
return rc;
}
@ -6580,6 +6669,7 @@ static int ha_connect_finish_peer(reconnect_ctx_t *rc_ctx)
if (rc < 0) {
ep->data = NULL;
endpoint_fini(ep);
free(ep);
free(sctx);
return rc;
}
@ -6675,6 +6765,7 @@ static void ha_connect_to_peer(cluster_t *c, int peer_idx)
* Tear down our duplicate prep; the next tick retries cleanly. */
rc_ctx->ep->data = NULL;
endpoint_fini(rc_ctx->ep);
free(rc_ctx->ep); /* endpoint_get() calloc'd it; fini frees inner only */
free(rc_ctx->sctx);
free(rc_ctx);
atomic_store_explicit(&ha->reconnect_inflight[peer_idx], 0,
@ -6690,6 +6781,7 @@ static void ha_connect_to_peer(cluster_t *c, int peer_idx)
peer_idx, srv);
rc_ctx->ep->data = NULL;
endpoint_fini(rc_ctx->ep);
free(rc_ctx->ep); /* endpoint_get() calloc'd it; fini frees inner only */
free(rc_ctx->sctx);
free(rc_ctx);
atomic_store_explicit(&ha->reconnect_inflight[peer_idx], 0,

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

@ -1118,6 +1118,162 @@ static void test_reconnect_offload_dedups_per_peer(void)
printf("test_reconnect_offload_dedups_per_peer: PASS\n");
}
/* ── Tests 18–19: raft-drive-async Task 2.2 — inbound/outbound ownership ─────
*
* Acceptance criteria (Req 2.6, 4.4; design ReconnectOffload race resolution,
* ha-async-io.md connection ownership):
* - When a reconnect succeeds and registers a NEW outbound control channel
* for peer X, any PRE-EXISTING standby_endpoints entry for the SAME peer X
* (a promoted vote-ep or a stale prior channel that has not yet been torn
* down) must be resolved deterministically: exactly ONE live outbound
* channel per peer survives — no dual live connection, no orphaned/leaked
* endpoint, and the survivor is the freshly-connected one so no queued
* message is delivered down a dead duplicate (lost/reordered, Req 2.6).
*
* The collision: BOTH raft_cb_become_leader (promotes vote eps) and
* on_standby_connected (late-joiner reconnect) zarray_store into
* standby_endpoints keyed only by ep-POINTER identity — never by peer_idx.
* So two distinct eps for the same peer can both register → two state-push
* fan-out targets → exactly the bug-851 dual-channel hazard.
*
* Determinism: this is the ownership-DECISION logic (not a real socket-timing
* race); the real concurrent inbound-accept-vs-outbound-connect timing race is
* covered by the partition soak (task 4.1). Here we drive the registration
* point directly with two heap endpoints for the same peer_idx and assert the
* resolution. Heap eps + endpoint_init make the teardown production-faithful
* so LeakSanitizer would flag an orphaned/leaked evicted endpoint. */
extern int on_standby_connected_for_test(endpoint_t *ep);
/* Production teardown of an outbound standby channel (removes from
* standby_endpoints + repl, frees ep->data); exposed for test teardown. */
extern int standby_ep_err_handler_primary_for_test(endpoint_t *ep);
/* Test-observable counter: how many stale same-peer channels on_standby_connected
* evicted while resolving ownership (raft-drive-async Task 2.2). */
extern _Atomic int ha_standby_evicted_count_for_test;
/* Allocate + init a heap endpoint wired as a standby channel for peer_idx, the
* way ha_connect_prep_peer / a promoted vote-ep would leave it (production
* teardown via standby_ep_err_handler_primary frees ep->data + endpoint, so the
* evicted one must be reclaimed without a leak). */
static endpoint_t *own_make_standby_ep(ha_state_t *ha, cluster_t *c, int peer_idx)
{
endpoint_t *ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
endpoint_init(ep, AF_INET, SOCK_STREAM, 0, 0);
/* No live iomgr in this unit fixture: leave ep->iomgr NULL so the eviction
* path skips iomgr_unmonitor (it would epoll_ctl on a bogus fd). The
* ownership-resolution logic under test (array + replicator) is unaffected. */
ep->iomgr = NULL;
standby_ep_ctx_build_for_test(ep, ha, c, peer_idx);
return ep;
}
static void own_setup_primary(cluster_t *c, myconf_t *conf)
{
memset(c, 0, sizeof(*c));
memset(conf, 0, sizeof(*conf));
init_ha_primary_with_standby_eps(&c->ha);
conf->peers_cfg.candidates_count = 3;
for (int i = 0; i < 3; 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;
}
/*
* Test 18 (Req 2.6/4.4): registering a NEW outbound channel for a peer that
* ALREADY has a live standby_endpoints entry (different ep, same peer_idx)
* resolves to exactly ONE live channel — the new one — and tears the stale one
* down (no dual connection, no leak).
*
* RED (before the resolution lands): on_standby_connected only de-dups by ep
* pointer, so both eps remain → count == 2 (dual live connection).
*/
static void test_reconnect_resolves_dual_channel_same_peer(void)
{
cluster_t c;
myconf_t conf;
own_setup_primary(&c, &conf);
atomic_store(&ha_standby_evicted_count_for_test, 0);
/* Stale prior channel for peer 2 already registered (e.g. a promoted
* vote-ep whose RDHUP has not yet been processed). */
endpoint_t *stale = own_make_standby_ep(&c.ha, &c, 2);
zarray_store(c.ha.standby_endpoints, stale);
TEST_ASSERT_EQUAL_INT(1, c.ha.standby_endpoints->count);
/* New reconnect to the SAME peer 2 completes and registers. */
endpoint_t *fresh = own_make_standby_ep(&c.ha, &c, 2);
on_standby_connected_for_test(fresh);
/* Exactly one live channel for peer 2 — the fresh one. */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, c.ha.standby_endpoints->count,
"Req 2.6/4.4: a reconnect for a peer with an existing channel must "
"leave exactly ONE live outbound channel (no dual connection)");
TEST_ASSERT_EQUAL_PTR_MESSAGE(fresh, zarray_get(c.ha.standby_endpoints, 0),
"Req 2.6: the survivor must be the freshly-connected channel so no "
"queued message is delivered down the dead duplicate");
TEST_ASSERT_EQUAL_INT_MESSAGE(1,
atomic_load(&ha_standby_evicted_count_for_test),
"the stale same-peer channel must be evicted exactly once");
/* fresh is owned by standby_endpoints; tear it down the production way so
* LSan sees the heap ep + ctx reclaimed (the err handler frees both). */
standby_ep_err_handler_primary_for_test(fresh);
endpoint_fini(fresh);
free(fresh);
fini_ha_primary_with_standby_eps(&c.ha);
printf("test_reconnect_resolves_dual_channel_same_peer: PASS\n");
}
/*
* Test 19 (Req 2.6): registering a NEW channel for a DIFFERENT peer does NOT
* evict an existing channel for another peer — resolution is strictly
* per-peer, so unrelated peers' channels and their queued messages are
* untouched.
*/
static void test_reconnect_preserves_other_peer_channel(void)
{
cluster_t c;
myconf_t conf;
own_setup_primary(&c, &conf);
atomic_store(&ha_standby_evicted_count_for_test, 0);
endpoint_t *peer1 = own_make_standby_ep(&c.ha, &c, 1);
zarray_store(c.ha.standby_endpoints, peer1);
endpoint_t *peer2 = own_make_standby_ep(&c.ha, &c, 2);
on_standby_connected_for_test(peer2);
TEST_ASSERT_EQUAL_INT_MESSAGE(2, c.ha.standby_endpoints->count,
"Req 2.6: a reconnect to one peer must not evict another peer's "
"live channel");
TEST_ASSERT_EQUAL_INT_MESSAGE(0,
atomic_load(&ha_standby_evicted_count_for_test),
"no eviction when the peer_idx differs");
standby_ep_err_handler_primary_for_test(peer1);
endpoint_fini(peer1);
free(peer1);
standby_ep_err_handler_primary_for_test(peer2);
endpoint_fini(peer2);
free(peer2);
fini_ha_primary_with_standby_eps(&c.ha);
printf("test_reconnect_preserves_other_peer_channel: PASS\n");
}
/* ── Suite entry point ──────────────────────────────────────────────────── */
void test_ha_liveness(void)
@ -1145,5 +1301,8 @@ void test_ha_liveness(void)
RUN_TEST(test_reconnect_offload_tick_returns_promptly);
RUN_TEST(test_reconnect_inline_tick_blocks_RED);
RUN_TEST(test_reconnect_offload_dedups_per_peer);
/* 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);
UNITY_END();
}