From 35d7e54a4a34209c47c6f6bad1931ccad0f0b3fc Mon Sep 17 00:00:00 2001 From: Groene AI <270696204+groeneai@users.noreply.github.com> Date: Tue, 18 Aug 2026 08:08:52 +0000 Subject: [PATCH 1/3] libuv adapter: do not let an attach overtake a queued poll removal The adapter defers poll mutation to the event loop thread through a FIFO, but natsLibuv_Attach did not follow that discipline in two ways. First, it wrote nle->socket and nle->events directly. On a reconnect that runs on the nats-io reconnect thread (_doReconnect -> _processConnInit -> evCbs.attach), those writes race the event loop thread's write in uvPollUpdate's close branch. ThreadSanitizer reports it as a data race on a 4-byte field of natsLibuvEvents. Second, it scheduled only when it was called off the loop thread, while Read/Write/Detach also schedule when events are already queued. A loop-thread attach could therefore run ahead of removals already in the list. Both let an attach install a new socket before an older, already queued READ removal is drained. _evStopPolling only requests the removals and then _processOpError starts the reconnect thread, so when the loop thread finally drains that removal, nle->events reaches 0 and uvPollUpdate takes the close branch on the socket the attach just installed: natsConnection_ProcessCloseEvent closes a live, freshly connected fd and sets nle->socket to invalid. The reconnect still reports success (nc->el.attached = true) but the connection never polls again. Carry the socket in the ATTACH event and assign nle->socket and nle->events inside uvAsyncAttach, so both fields are written only by the loop thread, and give the attach the same "or events are queued" guard as the other three callbacks. Ordering is then enforced by the queue instead of by timing, and the race disappears because the fields are no longer shared. No new lock is introduced; nle->lock still guards only head/tail. Reproduced with a standalone harness that links this header against libuv under ThreadSanitizer and drives the interleaving deterministically: before, 25/25 runs close the freshly attached fd and 23/25 concurrent runs report the data race in natsLibuv_Attach; after, 0 in both. (cherry picked from commit 37532c32464613665d7a9f65faf86f410b44eec7) --- src/adapters/libuv.h | 79 +++++++++++++++++++++++++++++++++----------- 1 file changed, 60 insertions(+), 19 deletions(-) diff --git a/src/adapters/libuv.h b/src/adapters/libuv.h index 4b3fc6f6f..e1bcf27de 100644 --- a/src/adapters/libuv.h +++ b/src/adapters/libuv.h @@ -41,6 +41,8 @@ typedef struct __natsLibuvEvent { int type; bool add; + // Only meaningful for NATS_LIBUV_ATTACH. + natsSock socket; struct __natsLibuvEvent *next; } natsLibuvEvent; @@ -107,18 +109,9 @@ natsLibuv_SetThreadLocalLoop(uv_loop_t *loop) } static natsStatus -uvScheduleToEventLoop(natsLibuvEvents *nle, int eventType, bool add) +uvEnqueueEvent(natsLibuvEvents *nle, natsLibuvEvent *newEvent) { - natsLibuvEvent *newEvent = NULL; - int res; - - newEvent = (natsLibuvEvent*) malloc(sizeof(natsLibuvEvent)); - if (newEvent == NULL) - return NATS_NO_MEMORY; - - newEvent->type = eventType; - newEvent->add = add; - newEvent->next = NULL; + int res; uv_mutex_lock(nle->lock); @@ -142,6 +135,50 @@ uvScheduleToEventLoop(natsLibuvEvents *nle, int eventType, bool add) return (res == 0 ? NATS_OK : NATS_ERR); } +static natsLibuvEvent* +uvNewEvent(int eventType, bool add) +{ + // Zero-initialized so that `socket` has a defined value for the event + // types that do not carry one. + natsLibuvEvent *newEvent = (natsLibuvEvent*) calloc(1, sizeof(natsLibuvEvent)); + + if (newEvent == NULL) + return NULL; + + newEvent->type = eventType; + newEvent->add = add; + newEvent->next = NULL; + + return newEvent; +} + +static natsStatus +uvScheduleToEventLoop(natsLibuvEvents *nle, int eventType, bool add) +{ + natsLibuvEvent *newEvent = uvNewEvent(eventType, add); + + if (newEvent == NULL) + return NATS_NO_MEMORY; + + return uvEnqueueEvent(nle, newEvent); +} + +// The socket travels in the event so that `nle` is not written here. Callers +// pass their own socket argument; reading `nle->socket` off the event loop +// thread is not allowed. +static natsStatus +uvScheduleAttachToEventLoop(natsLibuvEvents *nle, natsSock socket) +{ + natsLibuvEvent *newEvent = uvNewEvent(NATS_LIBUV_ATTACH, true); + + if (newEvent == NULL) + return NATS_NO_MEMORY; + + newEvent->socket = socket; + + return uvEnqueueEvent(nle, newEvent); +} + static void natsLibuvPoll(uv_poll_t* handle, int status, int events) { @@ -204,11 +241,16 @@ uvPollUpdate(natsLibuvEvents *nle, int eventType, bool add) return NATS_OK; } +// Runs on the event loop thread only, which is what makes `nle->socket` and +// `nle->events` single-threaded. static natsStatus -uvAsyncAttach(natsLibuvEvents *nle) +uvAsyncAttach(natsLibuvEvents *nle, natsSock socket) { natsStatus s = NATS_OK; + nle->socket = socket; + nle->events = UV_READABLE; + // Even when this is a reconnect, previous nle->handle has already been // set to NULL (and the memory has or will be freed in uvHandleClosedCb), // so recreate now. @@ -303,7 +345,7 @@ uvAsyncCb(uv_async_t *handle) { case NATS_LIBUV_ATTACH: { - s = uvAsyncAttach(nle); + s = uvAsyncAttach(nle, event->socket); break; } case NATS_LIBUV_READ: @@ -404,13 +446,12 @@ natsLibuv_Attach(void **userData, void *loop, natsConnection *nc, natsSock socke if (s == NATS_OK) { - nle->socket = socket; - nle->events = UV_READABLE; - - if (sched) - s = uvScheduleToEventLoop(nle, NATS_LIBUV_ATTACH, true); + // See comment in natsLibuvRead. Ordering matters here too: a pending + // removal must reach uvPollUpdate before this socket is installed. + if (sched || (nle->head != NULL)) + s = uvScheduleAttachToEventLoop(nle, socket); else - s = uvAsyncAttach(nle); + s = uvAsyncAttach(nle, socket); } if (s == NATS_OK) From d5c797c9f0c53ffe3752e2a9b5ee5901cb85d9e7 Mon Sep 17 00:00:00 2001 From: Alexey Milovidov Date: Sun, 13 Sep 2026 20:39:20 +0000 Subject: [PATCH 2/3] Do not close an already released poll handle in the libuv adapter A `NATS` connection whose credentials the server rejects on two consecutive reconnect attempts is closed for good by `_processAuthError`, and closing it crashes the process: uv_close contrib/libuv/src/unix/core.c:153 uvAsyncCb src/adapters/libuv.h:301 uv__async_io uv__io_poll uv_run `_evStopPolling` runs once per reconnect attempt and once more from `_close` when the library gives up, because `_close` only checks `nc->el.attached`, which `_evStopPolling` does not clear. Each run asks the adapter to remove the write and then the read event, and the removal which brings `nle->events` to zero closes the poll handle and sets `nle->handle` to `NULL`. Every further removal then reaches the unguarded `uv_close((uv_handle_t*) nle->handle, ...)` with a `NULL` handle, and `uv_close` faults reading the handle's type field. Guard `uvPollUpdate` against a handle which is already gone, and do not ask a connection which is not being polled to stop polling: `_evStopPolling` hands the socket close to the adapter, which by then only knows the socket it was given when it was attached, so a second run also leaked the socket of the last reconnect attempt. Reported as https://github.com/ClickHouse/ClickHouse/issues/118449 (cherry picked from commit 3e3a3f10572ed9ff1b7ad17873881877cb62f301) --- src/adapters/libuv.h | 8 ++++++++ src/conn.c | 9 +++++++-- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/src/adapters/libuv.h b/src/adapters/libuv.h index e1bcf27de..b0ef26909 100644 --- a/src/adapters/libuv.h +++ b/src/adapters/libuv.h @@ -224,6 +224,14 @@ uvPollUpdate(natsLibuvEvents *nle, int eventType, bool add) nle->events &= ~UV_WRITABLE; } + // The poll handle is released as soon as both events have been removed, and further + // events can still be delivered for it: the library asks to stop polling once per + // reconnect attempt and once more when it gives up and closes the connection, and the + // removals of the read and of the write event are two separate events. There is nothing + // left to poll on or to close, so both calls below would dereference a NULL handle. + if (nle->handle == NULL) + return NATS_OK; + if (nle->events) { int res = uv_poll_start(nle->handle, nle->events, natsLibuvPoll); diff --git a/src/conn.c b/src/conn.c index ed694e04d..58c6e517f 100644 --- a/src/conn.c +++ b/src/conn.c @@ -2846,8 +2846,13 @@ _close(natsConnection *nc, natsConnStatus status, bool fromPublicClose, bool doC // one doing it. if (ttj.readLoop == NULL) { - // If event loop attached, stop polling... - if (nc->el.attached) + // If event loop attached and still polling, stop polling... + // `_evStopPolling()` has already run for a connection which was reconnecting, + // and it leaves the actual socket close to the event loop adapter, which has + // released its handle by then. Asking a second time would leave the socket of + // the last reconnect attempt open forever, because the adapter only knows the + // socket it was given when it was attached. + if (nc->el.attached && nc->sockCtx.useEventLoop) { // This will take care of invalidating the socket and clear SSL, // but the actual socket close will be done from the event loop From b360a6e540a4dc5f5700ef582fcb9a3492076ecd Mon Sep 17 00:00:00 2001 From: Alexey Milovidov Date: Tue, 15 Sep 2026 13:58:35 +0000 Subject: [PATCH 3/3] Retry poll() on EINTR when waiting for a socket A signal delivered to the thread interrupts poll() with EINTR. natsSock_WaitReady treated it as a socket error, so a connection attempt made by an application with a sampling profiler (which sends a signal to every thread) failed spuriously with "poll error: 4". Wait again for whatever is left of the deadline instead. Co-Authored-By: Claude Fable 5.1 (cherry picked from commit b88704dfeb55e0f1dee165b9c1daccb90d5816d4) --- src/unix/sock.c | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/src/unix/sock.c b/src/unix/sock.c index 4bfcfa39c..16471feea 100644 --- a/src/unix/sock.c +++ b/src/unix/sock.c @@ -48,10 +48,18 @@ natsSock_WaitReady(int waitMode, natsSockCtx *ctx) abort(); } - if (deadline != NULL) - timeout = natsDeadline_GetTimeout(deadline); + do + { + if (deadline != NULL) + timeout = natsDeadline_GetTimeout(deadline); + + res = poll(&pfd, 1, timeout); + } + // A signal delivered to the thread (e.g. the sampling profiler of the host + // application) interrupts poll() with EINTR. That is not a socket error: + // wait again for whatever is left of the deadline. + while ((res == NATS_SOCK_ERROR) && (NATS_SOCK_GET_ERROR == EINTR)); - res = poll(&pfd, 1, timeout); if (res == NATS_SOCK_ERROR) return nats_setError(NATS_IO_ERROR, "poll error: %d", NATS_SOCK_GET_ERROR); else if (res == 0)