From e03ab9bd722bfa38be4a220bf55de2d61441a0a6 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 26 Jun 2026 09:54:10 +0000 Subject: [PATCH] feat(raft-drive-async): resolve inbound/outbound channel ownership on reconnect (task 2.2) 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. --- src/controller/ha.c | 108 ++++++++++++- src/controller/testsuite/test_ha_liveness.c | 159 ++++++++++++++++++++ 2 files changed, 259 insertions(+), 8 deletions(-) diff --git a/src/controller/ha.c b/src/controller/ha.c index ef61e52..2c19efa 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -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) { - ep->data = NULL; - endpoint_fini(ep); - free(sctx); - } + /* 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, diff --git a/src/controller/testsuite/test_ha_liveness.c b/src/controller/testsuite/test_ha_liveness.c index dffa47f..4e3a7fd 100644 --- a/src/controller/testsuite/test_ha_liveness.c +++ b/src/controller/testsuite/test_ha_liveness.c @@ -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(); }