Repository: vert-x3/vertx-web
Issue: Race condition when using an event bus bridge hook (events handler), between socket closed event and other events

### Version

* vert.x core: 3.7.1 (should apply to 3.8.1, master)
* vert.x web: 3.7.1 (should apply to 3.8.1, master)

### Context

We started to see that some messages were not delivered to the same event bus (bridge) address. After multiple re-connections, a client can't listen anymore to some particular address as it receives nothing or receives too few messages out of all the messages sent. The server needs to be restarted.

When an event bus bridge hook (events handler) is used, it seems like there's a race condition where io.vertx.ext.web.handler.sockjs.impl.EventBusBridgeImpl.java may call handleSocketClosed while we're still in internalHandleRegister (or unregister, sendorpub). This can happen before checkCallHook's okAction is run as we're still waiting for the hook (event handlers) to accept the event...

### Do you have a reproducer?

See steps below

### Steps to reproduce

We reproduced this with custom tools but I think this scenario is sufficient to reproduce it:

1. ... Set an event bus bridge events handler
2. ... Intercept "REGISTER" event and wait enough time before accepting the event
3. ... Meanwhile, disconnect the client
4. ... A MessageConsumer is created for the given address although socket is already closed and any message sent to this address might now land into this MessageConsumer associated to nothing

### Extra

Should apply to 3.8.1 and master as nothing was changed that could fix this issue in this file (EventBusBridgeImpl.java).

We fixed this internally by just checking for
`if (!sockInfos.containsKey(sock)) return;`
at the beginning of checkCallHook's okAction handlers.

It could probably explain https://github.com/vert-x3/vertx-web/issues/1327: the message consumer's handler references the closed socket when the race condition issue happens while a REGISTER event is being handled.

Changed files:
vertx-web/src/main/java/io/vertx/ext/web/handler/sockjs/impl/EventBusBridgeImpl.java

diff --git a/vertx-web/src/main/java/io/vertx/ext/web/handler/sockjs/impl/EventBusBridgeImpl.java b/vertx-web/src/main/java/io/vertx/ext/web/handler/sockjs/impl/EventBusBridgeImpl.java
index 346b45b8e..51a98c417 100644
--- a/vertx-web/src/main/java/io/vertx/ext/web/handler/sockjs/impl/EventBusBridgeImpl.java
+++ b/vertx-web/src/main/java/io/vertx/ext/web/handler/sockjs/impl/EventBusBridgeImpl.java
@@ -97,8 +97,11 @@ public class EventBusBridgeImpl implements Handler<SockJSSocket> {
     // On close unregister any handlers that haven't been unregistered
     registrations.forEach((key, value) -> {
       value.unregister();
-      checkCallHook(() -> new BridgeEventImpl(BridgeEventType.UNREGISTER,
-        new JsonObject().put("type", "unregister").put("address", value.address()), sock), null, null);
+      checkCallHook(() ->
+        new BridgeEventImpl(
+          BridgeEventType.UNREGISTER,
+          new JsonObject().put("type", "unregister").put("address", value.address()),
+          sock));
     });
 
     SockInfo info = sockInfos.remove(sock);
@@ -109,8 +112,7 @@ public class EventBusBridgeImpl implements Handler<SockJSSocket> {
       }
     }
 
-    checkCallHook(() -> new BridgeEventImpl(BridgeEventType.SOCKET_CLOSED, null, sock),
-      null, null);
+    checkCallHook(() -> new BridgeEventImpl(BridgeEventType.SOCKET_CLOSED, null, sock));
   }
 
   private void handleSocketData(SockJSSocket sock, Buffer data, Map<String, MessageConsumer<?>> registrations) {
@@ -158,19 +160,35 @@ public class EventBusBridgeImpl implements Handler<SockJSSocket> {
 
   }
 
+  private void checkCallHook(Supplier<BridgeEventImpl> eventSupplier) {
+    checkCallHook(eventSupplier, null, null);
+  }
+
   private void checkCallHook(Supplier<BridgeEventImpl> eventSupplier, Runnable okAction, Runnable rejectAction) {
     if (bridgeEventHandler == null) {
       if (okAction != null) {
         okAction.run();
       }
     } else {
-      BridgeEventImpl event = eventSupplier.get();
+      final BridgeEventImpl event = eventSupplier.get();
+      final boolean before = sockInfos.containsKey(event.socket());
       bridgeEventHandler.handle(event);
-      event.future().onComplete(res -> {
-        if (res.succeeded()) {
-          if (res.result()) {
-            if (okAction != null) {
-              okAction.run();
+      event.future()
+        .onFailure(err -> LOG.error("Failure in bridge event handler", err))
+        .onSuccess(ok -> {
+          if (ok) {
+            final boolean after = sockInfos.containsKey(event.socket());
+            if (before != after) {
+              // even though the event check is valid, the socket info isn't valid anymore
+              if (rejectAction != null) {
+                rejectAction.run();
+              } else {
+                LOG.debug("SockJSSocket state change prevented send or pub");
+              }
+            } else {
+              if (okAction != null) {
+                okAction.run();
+              }
             }
           } else {
             if (rejectAction != null) {
@@ -179,10 +197,7 @@ public class EventBusBridgeImpl implements Handler<SockJSSocket> {
               LOG.debug("Bridge handler prevented send or pub");
             }
           }
-        } else {
-          LOG.error("Failure in bridge event handler", res.cause());
-        }
-      });
+        });
     }
   }
 
@@ -261,7 +276,7 @@ public class EventBusBridgeImpl implements Handler<SockJSSocket> {
           registrations.put(address, reg);
           info.handlerCount++;
           // Notify registration completed
-          checkCallHook(() -> new BridgeEventImpl(BridgeEventType.REGISTERED, rawMsg, sock), null, null);
+          checkCallHook(() -> new BridgeEventImpl(BridgeEventType.REGISTERED, rawMsg, sock));
         } else {
           // inbound match failed
           if (debug) {
@@ -306,7 +321,7 @@ public class EventBusBridgeImpl implements Handler<SockJSSocket> {
     if (info != null) {
       info.pingInfo.lastPing = System.currentTimeMillis();
       // Trigger an event to allow custom behavior after updating lastPing
-      checkCallHook(() -> new BridgeEventImpl(BridgeEventType.SOCKET_PING, null, sock), null, null);
+      checkCallHook(() -> new BridgeEventImpl(BridgeEventType.SOCKET_PING, null, sock));
     }
   }