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

If the user destroys the connection while the read event was in progress
it could be possible for the library to crash trying to process a read
event.

We need to immediately stop polling when the read event is removed.

Changing also the way we retain the connection: we will now retain
on the very first attach, and the adapters will have to call
`natsConnection_Destroy()` when the event loop is fully detached.

Resolves #888

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>

Author: kozlovic

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 @@ _processConnInit(natsConnection *nc)
             // 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 @@ natsConnection_ProcessReadEvent(natsConnection *nc)
         }
     }
 
-    _retain(nc);
-
     buffer = nc->el.buffer;
     size   = nc->opts->ioBufSize;
 
@@ -4184,8 +4192,6 @@ natsConnection_ProcessReadEvent(natsConnection *nc)
 
     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;