From 81ea5e8eb81a34b5acce6b889b9e6b8162fff09d Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Sat, 30 May 2026 14:59:54 +0000 Subject: [PATCH] fix(election-iomgr-refactor): integration bug fixes from 6.3 cluster test - netio.c: skip SSL_new in endpoint_connect if ep->ssl already set (pre-created client ctx) - ha.c: pre-create client SSL before endpoint_connect to avoid server-ctx handshake failure - ha.c: fix double-free of vctx in endpoint_connect error path (let endpoint_fini own ep->data) - ha.c: add endpoint_set_peer() call after endpoint_init for vote endpoints - ha.c: fix endpoint_init args to AF_INET/SOCK_STREAM for outgoing vote sockets - ha.c: wrap handle_request_vote voted_for check in ha->lock to prevent split-brain - ha.c: skip immediate vote-deny for ECONNREFUSED in production; let election timeout handle - delay.c: fix task_cmp signature to match ordq_cmp_fn_t (3 args: ctx, arg1, arg2) --- src/controller/ha.c | 76 +++++++++++++++++++++------------------------ src/lib/delay.c | 3 +- src/lib/netio.c | 9 ++++-- 3 files changed, 44 insertions(+), 44 deletions(-) diff --git a/src/controller/ha.c b/src/controller/ha.c index 874a3da4..10639245 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -885,11 +885,16 @@ int handle_request_vote(ha_state_t *ha, if (in_catch_up_flag) return 0; + /* Protect voted_for check+set from concurrent vote requests on different vtasks */ + pthread_mutex_lock(&ha->lock); + uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire); /* Req 2.2: stale term → deny, do not update any state */ - if (term < my_term) + if (term < my_term) { + pthread_mutex_unlock(&ha->lock); return 0; + } /* Req 2.3: higher term → advance term, clear prior vote record */ if (term > my_term) { @@ -901,23 +906,31 @@ int handle_request_vote(ha_state_t *ha, /* Req 2.6: already voted for a different candidate in this term → deny */ if (ha->voted_for_term == term && ha->voted_for_name[0] != '\0' && - strcmp(ha->voted_for_name, candidate_id) != 0) + strcmp(ha->voted_for_name, candidate_id) != 0) { + pthread_mutex_unlock(&ha->lock); return 0; + } /* Req 2.4: freshness gate * Deny if candidate has a lower config_tag, or equal config_tag with lower LSN. * Grant if candidate is strictly fresher or equally fresh. */ - if (config_tag < my_config_tag) + if (config_tag < my_config_tag) { + pthread_mutex_unlock(&ha->lock); return 0; - if (config_tag == my_config_tag && last_lsn < my_last_lsn) + } + if (config_tag == my_config_tag && last_lsn < my_last_lsn) { + pthread_mutex_unlock(&ha->lock); return 0; + } /* Grant the vote: record voted_for and persist before responding (Req 9.1) */ ha->voted_for_term = term; strncpy(ha->voted_for_name, candidate_id, sizeof(ha->voted_for_name) - 1); ha->voted_for_name[sizeof(ha->voted_for_name) - 1] = '\0'; + pthread_mutex_unlock(&ha->lock); + /* workdir is not available here — persist with NULL (test mode or caller * wraps this function with workdir before sending the network reply). * Production dispatch path calls ha_persist_term(ha, workdir) explicitly @@ -1889,58 +1902,41 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c, continue; } - /* Initialize endpoint: client-side SSL, no listener (fd=-1 until connect) */ - endpoint_init(ep, -1, 0, 0, 0); + /* Initialize endpoint: AF_INET/SOCK_STREAM client socket for outgoing vote */ + endpoint_init(ep, AF_INET, SOCK_STREAM, 0, 0); + if (c) endpoint_set_peer(ep, c->myhost); ep->data = vctx; + if (c) ep->iomgr = &c->iomgr; - /* Wire up SSL context (client method for outgoing vote connections) */ - if (ha->client_ssl_ctx && c) { - /* SSL_CTX is set on the iomgr; endpoint_connect will use it */ - ep->iomgr = &c->iomgr; - /* Set client SSL context on the endpoint's SSL object after connect */ - /* endpoint_connect creates the SSL object from iomgr->ssl_ctx; - * for outgoing connections we need TLS_client_method context. - * Store client_ssl_ctx reference; the connect path will use it. */ - } else if (c) { - ep->iomgr = &c->iomgr; + /* Pre-create client SSL BEFORE endpoint_connect so netio uses it directly. + * endpoint_connect skips SSL_new when ep->ssl is already set (Req 2.2). */ + if (ha->client_ssl_ctx) { + SSL_CTX *client_ctx = (SSL_CTX *)ha->client_ssl_ctx; + ep->ssl = SSL_new(client_ctx); + if (ep->ssl) + SSL_set_connect_state(ep->ssl); } /* Set callbacks */ endpoint_set_on_connected(ep, on_vote_connected); endpoint_set_err_handler(ep, vote_ep_err_handler); - /* Connect asynchronously */ + /* Connect (blocking TCP + async SSL via iomgr after) */ int rc = endpoint_connect(ep, peer->hostname, peer->port); if (rc < 0) { WARN("ha_start_election_async: endpoint_connect failed for peer %d (%s:%d): %d", i, peer->hostname, peer->port, rc); - free(vctx); + /* endpoint_fini frees ep->data (vctx) and ep->ssl — do not double-free */ endpoint_fini(ep); - /* Count as denied */ - handle_request_vote_reply(ha, ctx, i, cur_term, 0, 0, 0); + /* In production mode (c != NULL), let the election timeout handle unreachable + * peers — do NOT count as immediate denial, or a startup race (ECONNREFUSED + * before peers are listening) collapses the election before the timeout fires. + * In test mode (c == NULL), count as denied so the election can resolve. */ + if (!c) + handle_request_vote_reply(ha, ctx, i, cur_term, 0, 0, 0); continue; } - /* Fix 3 (HIGH): endpoint_connect creates the SSL object from iomgr->ssl_ctx - * (server context). For outgoing vote connections we need TLS_client_method. - * After connect, replace the SSL object with one from client_ssl_ctx and - * explicitly set connect (client) state so the handshake goes client-side. - * The fd is already set on the old ssl object — re-set it on the new one. */ - if (ha->client_ssl_ctx && ep->ssl) { - SSL_CTX *client_ctx = (SSL_CTX *)ha->client_ssl_ctx; - SSL *new_ssl = SSL_new(client_ctx); - if (new_ssl) { - int ep_fd = SSL_get_fd(ep->ssl); - SSL_free(ep->ssl); - ep->ssl = new_ssl; - SSL_set_fd(ep->ssl, ep_fd); - SSL_set_connect_state(ep->ssl); - } - /* If SSL_new fails, fall through with the original SSL object; - * worst case the handshake uses server context on client side, - * which will fail and trigger vote_ep_err_handler (peer denied). */ - } - /* Register with iomgr for async I/O */ if (c) { rc = iomgr_monitor(ep); diff --git a/src/lib/delay.c b/src/lib/delay.c index 001fb690..d11629e7 100644 --- a/src/lib/delay.c +++ b/src/lib/delay.c @@ -94,8 +94,9 @@ static void* delay_worker(void* args) return NULL; } -static int task_cmp(void* arg1, void* arg2) +static int task_cmp(void* ctx, void* arg1, void* arg2) { + (void)ctx; delay_task_t* t1 = (delay_task_t*)arg1; delay_task_t* t2 = (delay_task_t*)arg2; time_t k1 = t1->nextrun; diff --git a/src/lib/netio.c b/src/lib/netio.c index b67305d8..43c00113 100644 --- a/src/lib/netio.c +++ b/src/lib/netio.c @@ -665,12 +665,15 @@ int endpoint_connect(endpoint_t* ep, char* host, int port) return rc; } - ep->ssl = SSL_new(iomgr->ssl_ctx); + /* If ep->ssl is already set (e.g. pre-created with client context for + * outgoing connections), use it; otherwise create from iomgr->ssl_ctx. */ + if (ep->ssl == NULL) { + ep->ssl = SSL_new(iomgr->ssl_ctx); + SSL_set_connect_state(ep->ssl); + } SSL_set_fd(ep->ssl, ep->fd); - SSL_set_connect_state(ep->ssl); - set_fd_nonblocking(ep->fd); ep->flags = 0;