[FIXED] EventLoop: Libuv may crash if connection destroyed while consuming

src/adapters/libevent.h

@@ -240,6 +240,8 @@ _freeCb(evutil_socket_t ignoredSocket, short ignoredEvent, void *arg)
         event_active(nle->keepActive, 0, 0);
         event_free(nle->keepActive);
     }
+    // This will release the connection that is retained by the library on the first attach.
+    natsConnection_Destroy(nle->nc);
     free(nle);
 }
 

src/adapters/libuv.h

@@ -250,6 +250,8 @@ uvFinalCloseCb(uv_handle_t* handle)
     free(nle->scheduler);
     uv_mutex_destroy(nle->lock);
     free(nle->lock);
+    // This will release the connection that is retained by the library on the first attach.
+    natsConnection_Destroy(nle->nc);
     free(nle);
 }
 
@@ -421,6 +423,10 @@ natsLibuv_Read(void *userData, bool add)
     natsStatus      s    = NATS_OK;
     bool            sched;
 
+    // If we remove, first stop polling immediately, then proceed as usual.
+    if (!add)
+        uv_poll_stop(nle->handle);
+
     sched = ((uv_key_get(&uvLoopThreadKey) != nle->loop) ? true : false);
 
     // If this call is made from a different thread than the event loop's

src/conn.c

@@ -2013,18 +2013,28 @@
             // event just after this call returns.
             nc->sockCtx.useEventLoop = true;
 
+            // For the very first attach, we will retain the connection.
+            if (!nc->el.retained)
+                _retain(nc);
+
             s = nc->opts->evCbs.attach(&(nc->el.data),
                                        nc->opts->evLoop,
                                        nc,
                                        (int) nc->sockCtx.fd);
             if (s == NATS_OK)
             {
                 nc->el.attached = true;
+                nc->el.retained = true;
             }
             else
             {
                 nc->sockCtx.useEventLoop = false;
 
+                // If this was the very first attach and we failed, release the connection
+                // to compensate for the retain above.
+                if (!nc->el.retained)
+                    _release(nc);
+
                 nats_setError(s,
                               "Error attaching to the event loop: %d - %s",
                               s, natsStatus_GetText(s));
@@ -4164,8 +4174,6 @@
         }
     }
 
-    _retain(nc);
-
     buffer = nc->el.buffer;
     size   = nc->opts->ioBufSize;
 
@@ -4184,8 +4192,6 @@
 
     if (s != NATS_OK)
         _processOpError(nc, s, false);
-
-    natsConn_release(nc);
 }
 
 void

src/natsp.h

@@ -771,6 +771,7 @@ struct __natsConnection
     {
         bool            attached;
         bool            writeAdded;
+        bool            retained;   // Will be set to true at the very first successful attach.
         void            *buffer;
         void            *data;
     } el;

test/test.c

@@ -20457,7 +20457,13 @@ _evLoopWrite(void *userData, bool add)
 static natsStatus
 _evLoopDetach(void *userData)
 {
-    struct threadArg *arg = (struct threadArg *) userData;
+    struct threadArg    *arg = (struct threadArg *) userData;
+    natsConnection      *nc  = NULL;
+
+    natsMutex_Lock(arg->m);
+    nc = arg->nc;
+    natsMutex_Unlock(arg->m);
+    natsConnection_Destroy(nc);
 
     natsMutex_Lock(arg->m);
     arg->detached++;
@@ -20542,6 +20548,8 @@ void test_EventLoop(void)
     IFOK(s, natsOptions_SetClosedCB(opts, _closedCb, (void*) &arg));
     testCond(s == NATS_OK);
 
+    arg.nc = nc;
+
     pid = _startServer("nats://127.0.0.1:4222", NULL, true);
     CHECK_SERVER_STARTED(pid);
 
@@ -20670,6 +20678,8 @@ void test_EventLoopRetryOnFailedConnect(void)
                                      _evLoopDetach));
     testCond(s == NATS_OK);
 
+    arg.nc = nc;
+
     test("Start event loop: ");
     natsMutex_Lock(arg.m);
     arg.sock = NATS_SOCK_INVALID;
@@ -20757,6 +20767,8 @@ void test_EventLoopTLS(void)
                                      _evLoopDetach));
     testCond(s == NATS_OK);
 
+    arg.nc = nc;
+
     test("Start server: ");
     pid = _startServer("nats://127.0.0.1:4443", "-config tls.conf", true);
     CHECK_SERVER_STARTED(pid);