Repository: eclipse-vertx/vert.x

Problem statement:
IllegalStateException on vertx.close: easy to reproduce
### Questions

Vert.x raises an unhandled IllegalStateException on every vertx.close.

```
java.lang.IllegalStateException: null
	at io.vertx.core.impl.CloseFuture.add(CloseFuture.java:56)
	at io.vertx.core.impl.ContextInternal.addCloseHook(ContextInternal.java:305)
	at io.vertx.core.impl.VertxImpl.scheduleTimeout(VertxImpl.java:521)
	at io.vertx.core.impl.VertxImpl.setTimer(VertxImpl.java:370)
	at io.vertx.core.http.impl.WebSocketImplBase.initiateConnectionCloseTimeout(WebSocketImplBase.java:566)
	at io.vertx.core.http.impl.ServerWebSocketImpl.lambda$close$0(ServerWebSocketImpl.java:149)
	at io.vertx.core.impl.future.FutureImpl$3.onSuccess(FutureImpl.java:141)
	at io.vertx.core.impl.future.FutureBase.lambda$emitSuccess$0(FutureBase.java:54)
	at io.vertx.core.impl.EventLoopContext.execute(EventLoopContext.java:87)
	at io.vertx.core.impl.DuplicatedContext.execute(DuplicatedContext.java:179)
	at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:51)
	at io.vertx.core.impl.future.FutureImpl.addListener(FutureImpl.java:196)
	at io.vertx.core.impl.future.PromiseImpl.addListener(PromiseImpl.java:23)
	at io.vertx.core.impl.future.FutureImpl.onComplete(FutureImpl.java:164)
	at io.vertx.core.impl.future.PromiseImpl.onComplete(PromiseImpl.java:23)
	at io.vertx.core.http.impl.ServerWebSocketImpl.close(ServerWebSocketImpl.java:145)
	at io.vertx.core.http.impl.WebSocketImplBase.close(WebSocketImplBase.java:142)
	at io.vertx.core.http.impl.Http1xConnectionBase.close(Http1xConnectionBase.java:95)
	at io.vertx.core.http.impl.Http1xServerConnection.close(Http1xServerConnection.java:72)
	at io.vertx.core.net.impl.ConnectionBase.close(ConnectionBase.java:131)
	at io.vertx.core.net.impl.VertxHandler.close(VertxHandler.java:158)
	at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:622)
	at io.netty.channel.AbstractChannelHandlerContext.access$1200(AbstractChannelHandlerContext.java:61)
	at io.netty.channel.AbstractChannelHandlerContext$11.run(AbstractChannelHandlerContext.java:611)
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:503)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:995)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:829)
```

### Version

4.3.1

### Context

Always reproduces.

### Do you have a reproducer?

Yes.

### Steps to reproduce

Execute the following Kotlin code. During the Thread.sleep you will see the mentioned exception.
```kotlin
val vertx = Vertx.vertx()
vertx.deployVerticle(object : AbstractVerticle() {
    override fun start(startPromise: Promise<Void>) {
        vertx.createHttpServer().webSocketHandler {
            it.accept()
        }.listen(8080)
        startPromise.complete()
    }
})
try {
    val client = HttpClient.newHttpClient()
    val listener = object : WebSocket.Listener {}
    client.newWebSocketBuilder().buildAsync(URI("ws://127.0.0.1:8080/"), listener).join()
} finally {
    vertx.close()
}
Thread.sleep(1000)
```
Wating for callbacks to finish doesn't change the exception.


Hints:

Patch:
diff --git a/src/main/java/io/vertx/core/http/impl/WebSocketImplBase.java b/src/main/java/io/vertx/core/http/impl/WebSocketImplBase.java
--- a/src/main/java/io/vertx/core/http/impl/WebSocketImplBase.java
+++ b/src/main/java/io/vertx/core/http/impl/WebSocketImplBase.java
@@ -563,7 +563,7 @@ void closeConnection() {
    */
   void initiateConnectionCloseTimeout(long timeoutMillis) {
     synchronized (conn) {
-      closeTimeoutID = context.owner().setTimer(timeoutMillis, id -> {
+      closeTimeoutID = context.setTimer(timeoutMillis, id -> {
         synchronized (conn) {
           closeTimeoutID = -1L;
         }
diff --git a/src/main/java/io/vertx/core/impl/AbstractContext.java b/src/main/java/io/vertx/core/impl/AbstractContext.java
--- a/src/main/java/io/vertx/core/impl/AbstractContext.java
+++ b/src/main/java/io/vertx/core/impl/AbstractContext.java
@@ -85,13 +85,13 @@ public final void endDispatch(ContextInternal previous) {
   @Override
   public long setPeriodic(long delay, Handler<Long> handler) {
     VertxImpl owner = (VertxImpl) owner();
-    return owner.scheduleTimeout(this, true, delay, TimeUnit.MILLISECONDS, handler);
+    return owner.scheduleTimeout(this, true, delay, TimeUnit.MILLISECONDS, false, handler);
   }
 
   @Override
   public long setTimer(long delay, Handler<Long> handler) {
     VertxImpl owner = (VertxImpl) owner();
-    return owner.scheduleTimeout(this, false, delay, TimeUnit.MILLISECONDS,handler);
+    return owner.scheduleTimeout(this, false, delay, TimeUnit.MILLISECONDS, false, handler);
   }
 
   @Override
diff --git a/src/main/java/io/vertx/core/impl/ContextInternal.java b/src/main/java/io/vertx/core/impl/ContextInternal.java
--- a/src/main/java/io/vertx/core/impl/ContextInternal.java
+++ b/src/main/java/io/vertx/core/impl/ContextInternal.java
@@ -277,12 +277,14 @@ default void runOnContext(Handler<Void> action) {
   ContextInternal duplicate();
 
   /**
-   * Like {@link Vertx#setPeriodic(long, Handler)} except the periodic timer will fire on this context.
+   * Like {@link Vertx#setPeriodic(long, Handler)} except the periodic timer will fire on this context and the
+   * timer will not be associated with the context close hook.
    */
   long setPeriodic(long delay, Handler<Long> handler);
 
   /**
-   * Like {@link Vertx#setTimer(long, Handler)} except the timer will fire on this context.
+   * Like {@link Vertx#setTimer(long, Handler)} except the timer will fire on this context and the timer
+   * will not be associated with the context close hook.
    */
   long setTimer(long delay, Handler<Long> handler);
 
diff --git a/src/main/java/io/vertx/core/impl/VertxImpl.java b/src/main/java/io/vertx/core/impl/VertxImpl.java
--- a/src/main/java/io/vertx/core/impl/VertxImpl.java
+++ b/src/main/java/io/vertx/core/impl/VertxImpl.java
@@ -358,7 +358,8 @@ public EventBus eventBus() {
   }
 
   public long setPeriodic(long delay, Handler<Long> handler) {
-    return scheduleTimeout(getOrCreateContext(), true, delay, TimeUnit.MILLISECONDS, handler);
+    ContextInternal ctx = getOrCreateContext();
+    return scheduleTimeout(ctx, true, delay, TimeUnit.MILLISECONDS, ctx.isDeployment(), handler);
   }
 
   @Override
@@ -367,7 +368,8 @@ public TimeoutStream periodicStream(long delay) {
   }
 
   public long setTimer(long delay, Handler<Long> handler) {
-    return scheduleTimeout(getOrCreateContext(), false, delay, TimeUnit.MILLISECONDS, handler);
+    ContextInternal ctx = getOrCreateContext();
+    return scheduleTimeout(ctx, false, delay, TimeUnit.MILLISECONDS, ctx.isDeployment(), handler);
   }
 
   @Override
@@ -510,14 +512,19 @@ public DnsClient createDnsClient(DnsClientOptions options) {
     return new DnsClientImpl(this, options);
   }
 
-  public long scheduleTimeout(ContextInternal context, boolean periodic, long delay, TimeUnit timeUnit, Handler<Long> handler) {
+  public long scheduleTimeout(ContextInternal context,
+                                              boolean periodic,
+                                              long delay,
+                                              TimeUnit timeUnit,
+                                              boolean addCloseHook,
+                                              Handler<Long> handler) {
     if (delay < 1) {
       throw new IllegalArgumentException("Cannot schedule a timer with delay < 1 ms");
     }
     long timerId = timeoutCounter.getAndIncrement();
     InternalTimerHandler task = new InternalTimerHandler(timerId, handler, periodic, context);
     timeouts.put(timerId, task);
-    if (context.isDeployment()) {
+    if (addCloseHook) {
       context.addCloseHook(task);
     }
     EventLoop el = context.nettyEventLoop();
@@ -526,7 +533,7 @@ public long scheduleTimeout(ContextInternal context, boolean periodic, long dela
     } else {
       task.future = el.schedule(task, delay, timeUnit);
     }
-    return timerId;
+    return task.id;
   }
 
   public AbstractContext getContext() {
@@ -869,18 +876,18 @@ public HAManager haManager() {
    * This class does not rely on the internal {@link #future} for the termination to handle the worker case
    * since the actual timer {@link #handler} execution is scheduled when the {@link #future} executes.
    */
-  private class InternalTimerHandler implements Handler<Void>, Closeable, Runnable {
+  class InternalTimerHandler implements Handler<Void>, Closeable, Runnable {
 
     private final Handler<Long> handler;
     private final boolean periodic;
-    private final long timerID;
+    private final long id;
     private final ContextInternal context;
     private final AtomicBoolean disposed = new AtomicBoolean();
     private volatile java.util.concurrent.Future<?> future;
 
-    InternalTimerHandler(long timerID, Handler<Long> runnable, boolean periodic, ContextInternal context) {
+    InternalTimerHandler(long id, Handler<Long> runnable, boolean periodic, ContextInternal context) {
       this.context = context;
-      this.timerID = timerID;
+      this.id = id;
       this.handler = runnable;
       this.periodic = periodic;
     }
@@ -893,12 +900,12 @@ public void run() {
     public void handle(Void v) {
       if (periodic) {
         if (!disposed.get()) {
-          handler.handle(timerID);
+          handler.handle(id);
         }
       } else if (disposed.compareAndSet(false, true)) {
-        timeouts.remove(timerID);
+        timeouts.remove(id);
         try {
-          handler.handle(timerID);
+          handler.handle(id);
         } finally {
           // Clean up after it's fired
           context.removeCloseHook(this);
@@ -918,7 +925,7 @@ private boolean cancel() {
 
     private boolean tryCancel() {
       if  (disposed.compareAndSet(false, true)) {
-        timeouts.remove(timerID);
+        timeouts.remove(id);
         future.cancel(false);
         return true;
       } else {
@@ -999,8 +1006,9 @@ public synchronized TimeoutStream handler(Handler<Long> handler) {
         if (id != null) {
           throw new IllegalStateException();
         }
+        ContextInternal ctx = getOrCreateContext();
         this.handler = handler;
-        id = scheduleTimeout(getOrCreateContext(), periodic, delay, TimeUnit.MILLISECONDS, this);
+        this.id = scheduleTimeout(ctx, periodic, delay, TimeUnit.MILLISECONDS, ctx.isDeployment(), this);
       } else {
         cancel();
       }


Test patch:
diff --git a/src/test/java/io/vertx/core/TimerTest.java b/src/test/java/io/vertx/core/TimerTest.java
--- a/src/test/java/io/vertx/core/TimerTest.java
+++ b/src/test/java/io/vertx/core/TimerTest.java
@@ -460,7 +460,7 @@ public void testRaceWhenTimerCreatedOutsideEventLoop() {
     for (int i = 0;i < numThreads;i++) {
       Thread th = new Thread(() -> {
         // We need something more aggressive than a millisecond for this test
-        ((VertxImpl)vertx).scheduleTimeout(((VertxImpl) vertx).getOrCreateContext(), false, 1, TimeUnit.NANOSECONDS, ignore -> {
+        ((VertxImpl)vertx).scheduleTimeout(((VertxImpl) vertx).getOrCreateContext(), false, 1, TimeUnit.NANOSECONDS, false, ignore -> {
           count.decrementAndGet();
         });
       });
@@ -469,4 +469,25 @@ public void testRaceWhenTimerCreatedOutsideEventLoop() {
     }
     waitUntil(() -> count.get() == 0);
   }
+
+  @Test
+  public void testContextTimer() {
+    waitFor(2);
+    vertx.deployVerticle(new AbstractVerticle() {
+      @Override
+      public void start() throws Exception {
+        ((ContextInternal)context).setTimer(1000, id -> {
+          complete();
+        });
+        context.runOnContext(v -> {
+          vertx.undeploy(context.deploymentID(), onSuccess(ar -> {
+            ((ContextInternal)context).setTimer(1, id -> {
+              complete();
+            });
+          }));
+        });
+      }
+    });
+    await();
+  }
 }