lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(libxce-failover): classify and propagate netio I/O errors on failover

Browse source 
Add io_classify_io_error and propagate endpoint/epoll errors through
endpoint_update_epoll and cbuf_get instead of swallowing them, so client
failover surfaces the real error rather than hanging. Adds netio arm-failure,
close-propagation, and request send-propagation tests.

Pre-existing uncommitted work, committed to establish a clean baseline before
ha-commit-durability implementation.
This commit is contained in:
Rongsong Shen 2026-06-18 04:17:07 +00:00
parent 749e194207
commit 80bb1d766e
7 changed files with 347 additions and 41 deletions
Download patch file Download diff file Expand all files Collapse all files

3
src/lib/meson.build
View file

@ -173,6 +173,9 @@ libxce_test_src = ['testsuite/test.c',
'testsuite/test_cluster_parse.c',
'testsuite/test_libxce_retry.c',
'testsuite/test_netio_on_connected.c',
'testsuite/test_netio_arm_failure.c',
'testsuite/test_netio_close_propagation.c',
'testsuite/test_request_send_propagation.c',
'testsuite/test_raft.c',
'testsuite/test_wal_wire.c',
'raft/raft_sim.c',

111
src/lib/netio.c
View file

@ -69,6 +69,41 @@ int io_fsm_start(io_fsm_t* fsm, io_packet_t* pkt)
return 0;
}
/*
* io_classify_io_error: map a failed SSL_read/SSL_write (io_rc <= 0) to an FSM
* step result, given SSL_get_error()'s value and the errno captured right after
* the SSL call.
*
* returns 0 → retriable would-block (WANT_READ/WANT_WRITE, or an
* EAGAIN/EWOULDBLOCK syscall): the FSM stays put and retries on
* the next readiness event.
* returns <0 → the peer closed or reset the connection: the caller MUST move
* the FSM to FSM_STATE_ERROR and propagate this so a blocked
* reply-waiter is woken. A clean EOF reports errno==0, and the
* old code returned `-errno == 0` here — making the close look
* like success, so handle_input never reached handle_error and a
* blocked cbuf_get() hung forever (bug-648). We therefore return
* -ECONNRESET whenever errno is 0.
*
* Pure (no SSL object, no endpoint state) so it is unit-testable.
*/
int io_classify_io_error(int ssl_err, int ecode, int io_rc)
{
switch (ssl_err) {
case SSL_ERROR_WANT_READ:
case SSL_ERROR_WANT_WRITE:
return 0;
case SSL_ERROR_SYSCALL:
if ((io_rc < 0) && ((ecode == EAGAIN) || (ecode == EWOULDBLOCK)))
return 0;
return (ecode != 0) ? -ecode : -ECONNRESET;
default:
/* SSL_ERROR_ZERO_RETURN (clean TLS close_notify) and any other SSL
* error: the session is closed — never report success. */
return (ecode != 0) ? -ecode : -ECONNRESET;
}
}
static int io_fsm_do(io_fsm_t* fsm, char* data, size_t len)
{
int rc = 0;
@ -87,32 +122,15 @@ static int io_fsm_do(io_fsm_t* fsm, char* data, size_t len)
}
if (rc <= 0) {
int err = 0;
int ecode = 0;
int ecode = errno;
int err = SSL_get_error(ep->ssl, rc);
ecode = errno;
err = SSL_get_error(ep->ssl, rc);
switch (err) {
case SSL_ERROR_WANT_READ:
case SSL_ERROR_WANT_WRITE:
rc = 0;
break;
case SSL_ERROR_SYSCALL:
if ((rc < 0) && ((ecode == EAGAIN) || (ecode == EWOULDBLOCK))) {
rc = 0;
} else {
fsm->state = FSM_STATE_ERROR;
rc = -errno;
}
if (rc == 0) {
/* we meet EOF */
fsm->state = FSM_STATE_ERROR;
}
break;
default:
/* Classify the failure. A negative result means the connection is
* closed/errored — move to ERROR and propagate (never return 0 on a
* close, or a blocked reply-waiter is never woken: bug-648). */
rc = io_classify_io_error(err, ecode, rc);
if (rc < 0) {
fsm->state = FSM_STATE_ERROR;
rc = -errno;
break;
}
return rc;
@ -411,7 +429,13 @@ static int handle_output(endpoint_t* ep)
ep->fsm_out.pkt = NULL;
endpoint_update_epoll(ep);
/* Re-arm (drop EPOLLOUT now the queue is drained). A failure here means
* the fd is no longer registered — surface it so io_handler runs the
* error path rather than leaving a silently-dead endpoint (bug-646). */
rc = endpoint_update_epoll(ep);
if (rc < 0) {
return rc;
}
}
return 0;
@ -558,27 +582,41 @@ static int endpoint_calc_flags(endpoint_t* ep)
static int endpoint_update_epoll(endpoint_t* ep)
{
int flags;
int rc = 0;
mylock_lock(&(ep->lock));
flags = endpoint_calc_flags(ep);
if (flags != ep->flags) {
if (ep->iomgr != NULL) {
int rc1;
int saved_errno;
mylock_lock(&(ep->iomgr->lock));
rc1 = iomgr_update_poll(ep->iomgr, ep, flags);
rc1 = iomgr_update_poll(ep->iomgr, ep, flags);
saved_errno = errno;
mylock_unlock(&(ep->iomgr->lock));
if (rc1 >= 0) {
ep->flags = flags;
} else {
WARN("rc1=%d < 0", rc1);
/* Do NOT swallow this: an unarmed poll interest silently
* strands all I/O on the endpoint — outbound bytes never flush
* and inbound replies never wake a blocked cbuf_get(), so a
* synchronous caller hangs forever while io_worker sleeps in
* epoll_wait(). EPOLL_CTL_MOD fails with ENOENT when the fd is
* no longer registered (e.g. after a failed reconnect).
* Surface the failure so the caller tears the endpoint down and
* the failover loop rotates instead of hanging (bug-646). */
rc = (saved_errno != 0) ? -saved_errno : -EIO;
WARN("endpoint_update_epoll: iomgr_update_poll failed "
"(rc1=%d errno=%d) — endpoint marked dead",
rc1, saved_errno);
}
}
}
mylock_unlock(&(ep->lock));
return 0;
return rc;
}
static void outq_item_killer(data_item_t* it)
@ -807,7 +845,22 @@ int endpoint_enq(endpoint_t* ep, data_item_t* it)
rc = dataq_enq(&(ep->outq), it);
mylock_unlock(&(ep->lock));
endpoint_update_epoll(ep);
/* Arm the output interest. If the poll registration cannot be updated
* (e.g. EPOLL_CTL_MOD on an fd no longer registered after a failed
* reconnect), the enqueued bytes will never flush and the endpoint is dead.
* Surface a RETRIABLE transport error so the caller propagates it (the
* request.c senders return it; the client failover loop classifies
* -ECONNRESET as retriable and rotates — bug-650).
*
* Do NOT invoke handle_error()/the err_handler here. endpoint_enq() runs on
* the *caller's* thread, which may already hold a lock the err_handler needs
* — e.g. the controller's primary_heartbeat_fn holds ha->lock and
* vote_ep_err_handler re-acquires it, self-deadlocking the controller
* (bug-652). The real teardown stays on the io_worker thread, which drives
* handle_error() off the next epoll error event holding no caller locks. */
if (endpoint_update_epoll(ep) < 0) {
return -ECONNRESET;
}
return rc;
}

24
src/lib/request.c
View file

@ -335,8 +335,8 @@ int xce_request(endpoint_t* ep, xce_request_t* req)
return rc;
}
endpoint_enq(ep, &(pkt->link));
return 0;
int enq_rc = endpoint_enq(ep, &(pkt->link));
return (enq_rc < 0) ? enq_rc : 0;
}
int xce_reply(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply)
@ -371,8 +371,8 @@ int xce_reply(endpoint_t* ep, xce_peerinfo_t* from, xce_reply_t* reply)
return rc;
}
endpoint_enq(ep, &(pkt->link));
return 0;
int enq_rc = endpoint_enq(ep, &(pkt->link));
return (enq_rc < 0) ? enq_rc : 0;
}
/*
@ -453,8 +453,8 @@ int xce_replconfigpush_req_raw(endpoint_t *ep, xce_peerinfo_t *from,
capn_write_mem(&c, (uint8_t *)pkt->data, pkt->sz, 0);
capn_free(&c);
endpoint_enq(ep, &(pkt->link));
return 0;
int enq_rc = endpoint_enq(ep, &(pkt->link));
return (enq_rc < 0) ? enq_rc : 0;
}
/* wal_entry_to_wire: fill a wire WalCatchupEntry from a C entry's header
@ -559,8 +559,8 @@ int xce_wal_batch_req_raw(endpoint_t *ep, wal_catchup_entry_t **entries,
return rc;
}
endpoint_enq(ep, &(pkt->link));
return 0;
int enq_rc = endpoint_enq(ep, &(pkt->link));
return (enq_rc < 0) ? enq_rc : 0;
}
/*
@ -644,8 +644,8 @@ int xce_snapshot_chunk_req_raw(endpoint_t *ep, const snapshot_transfer_t *chunk,
return rc;
}
endpoint_enq(ep, &(pkt->link));
return 0;
int enq_rc = endpoint_enq(ep, &(pkt->link));
return (enq_rc < 0) ? enq_rc : 0;
}
/*
@ -740,8 +740,8 @@ int xce_replconfig_reply_raw(endpoint_t *ep, xce_peerinfo_t *from,
capn_write_mem(&c, (uint8_t *)pkt->data, pkt->sz, 0);
capn_free(&c);
endpoint_enq(ep, &(pkt->link));
return 0;
int enq_rc = endpoint_enq(ep, &(pkt->link));
return (enq_rc < 0) ? enq_rc : 0;
}
/*

6
src/lib/testsuite/test.c
View file

@ -36,6 +36,9 @@ extern void test_cluster(void);
extern void test_cluster_parse(void);
extern void test_libxce_retry(void);
extern void test_netio_on_connected(void);
extern void test_netio_arm_failure(void);
extern void test_netio_close_propagation(void);
extern void test_request_send_propagation(void);
extern void test_raft(void);
extern void test_wal_wire(void);
@ -143,6 +146,9 @@ int main(int argc, char* argv[])
TEST(test_cluster_parse);
TEST(test_libxce_retry);
TEST(test_netio_on_connected);
TEST(test_netio_arm_failure);
TEST(test_netio_close_propagation);
TEST(test_request_send_propagation);
TEST(test_raft);
TEST(test_wal_wire);

95
src/lib/testsuite/test_netio_arm_failure.c Normal file
View file

@ -0,0 +1,95 @@
/*
* test_netio_arm_failure.c — libxce-failover: a poll-arm failure must fail LOUD.
*
* Root cause (bug-646): endpoint_update_epoll() swallowed iomgr_update_poll()
* (EPOLL_CTL_MOD) failures — it logged "rc1=-1 < 0" but returned 0 and left the
* poll interest unarmed. On the synchronous client send path (endpoint_enq),
* a failed arm meant the request bytes were never flushed, no reply ever
* arrived, and the caller blocked forever in cbuf_get() while a perfectly
* healthy io_worker slept in epoll_wait(). (EPOLL_CTL_MOD fails with ENOENT
* when the fd is no longer registered — e.g. after a failed reconnect.)
*
* Contract under test: when the poll interest cannot be armed, endpoint_enq()
* must (1) return a negative, RETRIABLE error so the caller propagates it
* (request.c senders return it; the client failover loop rotates — bug-650),
* and (2) must NOT synchronously invoke the endpoint's err_handler. endpoint_enq
* runs on the caller's thread, which may already hold a lock the err_handler
* needs (the controller's primary_heartbeat_fn holds ha->lock and
* vote_ep_err_handler re-acquires it → self-deadlock, bug-652). The real
* teardown stays on the io_worker thread, off the next epoll error event.
*/
#define _GNU_SOURCE
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
#include <unity/unity.h>
#include "netio.h"
/* ── err_handler invocation tracker ──────────────────────────────── */
static int g_err_count = 0;
static int err_handler_stub(void *ep)
{
(void)ep;
g_err_count++;
return 0;
}
/*
* Arming EPOLL_CTL_MOD on an fd that was never EPOLL_CTL_ADD'd fails with
* ENOENT — the exact post-failed-reconnect condition. endpoint_enq() must
* surface it (negative rc) WITHOUT calling the err_handler (no re-entrancy).
*/
static void test_enq_arm_failure_is_loud_no_errhandler(void)
{
iomgr_t iomgr;
memset(&iomgr, 0, sizeof(iomgr));
TEST_ASSERT_EQUAL_INT(0, iomgr_init(&iomgr));
endpoint_t *ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
/* A real socket fd that is NOT registered in iomgr's epoll: EPOLL_CTL_MOD
* on it fails with ENOENT, mirroring a send after a failed reconnect. */
ep->fd = socket(AF_INET, SOCK_STREAM, 0);
TEST_ASSERT_TRUE(ep->fd >= 0);
ep->iomgr = &iomgr;
ep->ssl_handshake_done = 1; /* past handshake: calc_flags arms EPOLLIN */
ep->flags = 0; /* force flags != calc_flags → iomgr_update_poll */
endpoint_set_err_handler(ep, (err_handler_t)err_handler_stub);
g_err_count = 0;
/* One packet on the outq so there is real outbound work to strand. The
* endpoint owns it after enq; endpoint_fini drains it via outq_item_killer. */
io_packet_t *pkt = (io_packet_t *)calloc(1, sizeof(*pkt));
TEST_ASSERT_NOT_NULL(pkt);
int rc = endpoint_enq(ep, &pkt->link);
/* Loud failure (negative, retriable) so the caller propagates and rotates —
* but the err_handler must NOT be invoked from this synchronous send path
* (re-entrancy would self-deadlock the controller, bug-652). */
TEST_ASSERT_TRUE(rc < 0);
TEST_ASSERT_EQUAL_INT(0, g_err_count);
endpoint_fini(ep);
free(ep);
close(iomgr.epoll_fd);
printf("test_enq_arm_failure_is_loud_no_errhandler: PASS\n");
}
/* ── suite entry point ───────────────────────────────────────────── */
void test_netio_arm_failure(void)
{
printf("\n>> netio poll-arm failure tests\n\n");
RUN_TEST(test_enq_arm_failure_is_loud_no_errhandler);
}

77
src/lib/testsuite/test_netio_close_propagation.c Normal file
View file

@ -0,0 +1,77 @@
/*
* test_netio_close_propagation.c — bug-648: a peer close/EOF must propagate as a
* negative error, never as rc==0.
*
* io_fsm_do() moved the read FSM to FSM_STATE_ERROR on a peer close but returned
* rc = -errno, which on a clean EOF (errno==0) is 0. handle_input() then read
* rc>=0, never called handle_error, and a blocked cbuf_get() reply-waiter was
* never woken. io_classify_io_error() is the extracted, pure decision; it must
* return <0 for every close/error case and 0 only for retriable would-block.
*/
#define _GNU_SOURCE
#include <errno.h>
#include <stdio.h>
#include <openssl/ssl.h>
#include <unity/unity.h>
/* Pure classifier defined (non-static) in netio.c, linked into this binary. */
extern int io_classify_io_error(int ssl_err, int ecode, int io_rc);
/* WANT_READ / WANT_WRITE are retriable would-block → rc==0, FSM stays put. */
static void test_wantread_writeready_is_retriable_zero(void)
{
TEST_ASSERT_EQUAL_INT(0, io_classify_io_error(SSL_ERROR_WANT_READ, 0, -1));
TEST_ASSERT_EQUAL_INT(0, io_classify_io_error(SSL_ERROR_WANT_WRITE, 0, -1));
printf("test_wantread_writeready_is_retriable_zero: PASS\n");
}
/* Non-blocking EAGAIN/EWOULDBLOCK syscall (io_rc<0) is retriable → rc==0. */
static void test_syscall_eagain_is_retriable_zero(void)
{
TEST_ASSERT_EQUAL_INT(0, io_classify_io_error(SSL_ERROR_SYSCALL, EAGAIN, -1));
TEST_ASSERT_EQUAL_INT(0,
io_classify_io_error(SSL_ERROR_SYSCALL, EWOULDBLOCK, -1));
printf("test_syscall_eagain_is_retriable_zero: PASS\n");
}
/*
* THE BUG: a clean EOF — SSL_read returns 0 with errno==0 — must yield a
* NEGATIVE error (not 0) so the close propagates and the waiter is woken.
*/
static void test_clean_eof_is_negative_not_zero(void)
{
int rc = io_classify_io_error(SSL_ERROR_SYSCALL, 0 /*errno*/, 0 /*io_rc EOF*/);
TEST_ASSERT_TRUE(rc < 0);
TEST_ASSERT_EQUAL_INT(-ECONNRESET, rc);
printf("test_clean_eof_is_negative_not_zero: PASS\n");
}
/* Clean TLS close_notify (ZERO_RETURN) with errno==0 must also be negative. */
static void test_zero_return_is_negative_not_zero(void)
{
int rc = io_classify_io_error(SSL_ERROR_ZERO_RETURN, 0, 0);
TEST_ASSERT_TRUE(rc < 0);
TEST_ASSERT_EQUAL_INT(-ECONNRESET, rc);
printf("test_zero_return_is_negative_not_zero: PASS\n");
}
/* A hard syscall error preserves its real errno (still negative). */
static void test_hard_syscall_error_preserves_errno(void)
{
TEST_ASSERT_EQUAL_INT(-ECONNRESET,
io_classify_io_error(SSL_ERROR_SYSCALL, ECONNRESET, -1));
printf("test_hard_syscall_error_preserves_errno: PASS\n");
}
void test_netio_close_propagation(void)
{
printf("\n>> netio peer-close propagation tests\n\n");
RUN_TEST(test_wantread_writeready_is_retriable_zero);
RUN_TEST(test_syscall_eagain_is_retriable_zero);
RUN_TEST(test_clean_eof_is_negative_not_zero);
RUN_TEST(test_zero_return_is_negative_not_zero);
RUN_TEST(test_hard_syscall_error_preserves_errno);
}

72
src/lib/testsuite/test_request_send_propagation.c Normal file
View file

@ -0,0 +1,72 @@
/*
* test_request_send_propagation.c — bug-650: xce_request() must PROPAGATE an
* endpoint_enq() send failure, not swallow it.
*
* xce_request() (and the other request.c senders) ignored endpoint_enq()'s
* return and always returned 0. So when the transport was dead — endpoint_enq()
* returns <0 and marks the endpoint dead (bug-646) — xce_request() still
* reported success. xce_jobsubmit() then fell through to a BLOCKING cbuf_get()
* and hung forever waiting for a reply to a request that was never transmitted
* (server idle, client fd unconnected). xce_request() must return the negative
* enq error so the failover loop rotates instead of blocking.
*/
#define _GNU_SOURCE
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
#include <unity/unity.h>
#include "netio.h"
#include "request.h"
/*
* A request that ENCODES successfully but whose send fails because the endpoint
* is dead (EPOLL_CTL_MOD on an fd that is not registered → ENOENT) must make
* xce_request() return a negative error, not 0.
*/
static void test_xce_request_propagates_send_failure(void)
{
iomgr_t iomgr;
memset(&iomgr, 0, sizeof(iomgr));
TEST_ASSERT_EQUAL_INT(0, iomgr_init(&iomgr));
endpoint_t *ep = endpoint_get();
TEST_ASSERT_NOT_NULL(ep);
ep->fd = socket(AF_INET, SOCK_STREAM, 0); /* valid fd, NOT in the epoll */
TEST_ASSERT_TRUE(ep->fd >= 0);
ep->iomgr = &iomgr;
ep->ssl_handshake_done = 1;
ep->flags = 0; /* force iomgr_update_poll MOD on the unregistered fd */
char nm[] = "tester";
char ky[] = "key";
xce_peerinfo_t peer = { .name = nm, .key = ky };
ep->peer = &peer;
xce_request_t req;
memset(&req, 0, sizeof(req));
req.opcode = OPCode_nop;
req.seq = 42;
int rc = xce_request(ep, &req);
/* Old code: always 0 (swallowed). Fixed: the negative enq error propagates. */
TEST_ASSERT_TRUE(rc < 0);
endpoint_fini(ep);
free(ep);
close(iomgr.epoll_fd);
printf("test_xce_request_propagates_send_failure: PASS\n");
}
void test_request_send_propagation(void)
{
printf("\n>> request send-failure propagation tests\n\n");
RUN_TEST(test_xce_request_propagates_send_failure);
}