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)
This commit is contained in:
parent
e875cae7ad
commit
81ea5e8eb8
3 changed files with 44 additions and 44 deletions
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue