diff --git a/src/lib/meson.build b/src/lib/meson.build index c0c48157..0e5401d6 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -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', diff --git a/src/lib/netio.c b/src/lib/netio.c index fa6f8a83..5cf5a545 100644 --- a/src/lib/netio.c +++ b/src/lib/netio.c @@ -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; } diff --git a/src/lib/request.c b/src/lib/request.c index 0d3924be..d1ef8dc1 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -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; } /* diff --git a/src/lib/testsuite/test.c b/src/lib/testsuite/test.c index 73516ef4..c688b5e8 100644 --- a/src/lib/testsuite/test.c +++ b/src/lib/testsuite/test.c @@ -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); diff --git a/src/lib/testsuite/test_netio_arm_failure.c b/src/lib/testsuite/test_netio_arm_failure.c new file mode 100644 index 00000000..c9981b89 --- /dev/null +++ b/src/lib/testsuite/test_netio_arm_failure.c @@ -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 +#include +#include +#include +#include +#include + +#include +#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); +} diff --git a/src/lib/testsuite/test_netio_close_propagation.c b/src/lib/testsuite/test_netio_close_propagation.c new file mode 100644 index 00000000..3928660e --- /dev/null +++ b/src/lib/testsuite/test_netio_close_propagation.c @@ -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 +#include + +#include +#include + +/* 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); +} diff --git a/src/lib/testsuite/test_request_send_propagation.c b/src/lib/testsuite/test_request_send_propagation.c new file mode 100644 index 00000000..002a5205 --- /dev/null +++ b/src/lib/testsuite/test_request_send_propagation.c @@ -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 +#include +#include +#include +#include +#include + +#include +#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); +}