Check invariant at the end of stress tests and fix issue where the in…

…correct capacity was redeemed when a connection is closed by the pool

src/main/java/io/vertx/core/http/impl/pool/Pool.java

@@ -120,6 +120,22 @@ public Pool(ConnectionProvider<C> connector,
     this.connectionRemoved = connectionRemoved;
   }
 
+  public synchronized int waitersInQueue() {
+    return waitersQueue.size();
+  }
+
+  public synchronized int waitersCount() {
+    return waitersCount;
+  }
+
+  public synchronized long weight() {
+    return weight;
+  }
+
+  public synchronized long capacity() {
+    return capacity;
+  }
+
   /**
    * Get a connection for a waiter asynchronously.
    *
@@ -318,22 +334,22 @@ private boolean deliverToWaiter(Holder<C> conn, Waiter<C> waiter) {
   // These methods assume to be called under synchronization
 
   private void recycleConnection(Holder<C> conn, int c, boolean closeable) {
-    long nc = conn.capacity + c;
-    if (nc > conn.concurrency) {
+    long newCapacity = conn.capacity + c;
+    if (newCapacity > conn.concurrency) {
       log.debug("Attempt to recycle a connection more than permitted");
       return;
     }
-    if (closeable && nc == conn.concurrency && waitersQueue.isEmpty()) {
+    if (closeable && newCapacity == conn.concurrency && waitersQueue.isEmpty()) {
       available.remove(conn);
-      capacity -= conn.concurrency;
+      capacity -= conn.capacity;
       conn.capacity = 0;
       connector.close(conn.connection);
     } else {
       capacity += c;
       if (conn.capacity == 0) {
         available.add(conn);
       }
-      conn.capacity = nc;
+      conn.capacity = newCapacity;
     }
   }
 

src/test/java/io/vertx/test/core/net/ConnectionPoolTest.java

@@ -41,7 +41,6 @@ class FakeConnectionManager {
     private final int queueMaxSize;
     private final int maxPoolSize;
     private Pool<FakeConnection> pool;
-    private int size;
     private Set<FakeConnection> active = new HashSet<>();
     private boolean closed = true;
     private int seq;
@@ -65,7 +64,7 @@ synchronized boolean contains(FakeConnection conn) {
     }
 
     synchronized int size() {
-      return size;
+      return active.size();
     }
 
     synchronized Pool<FakeConnection> pool() {
@@ -88,11 +87,9 @@ void getConnection(Waiter<FakeConnection> waiter) {
             }, (channel, conn) -> {
             synchronized (FakeConnectionManager.this) {
               active.add(conn);
-              size++;
             }
           }, (channel, conn) -> {
             synchronized (FakeConnectionManager.this) {
-              size--;
               active.remove(conn);
             }
           }
@@ -421,12 +418,12 @@ public boolean handleConnection(ContextInternal ctx, FakeConnection conn) throws
                 throw new Exception();
               } */ else {
                 vertx.setTimer(10, id -> {
-                  latch.countDown();
                   if (action < 15) {
                     conn.close();
                   } else {
                     conn.recycle();
                   }
+                  latch.countDown();
                 });
                 return true;
               }
@@ -449,6 +446,15 @@ public boolean handleConnection(ContextInternal ctx, FakeConnection conn) throws
         e.printStackTrace();
       }
     }
+
+    assertWaitUntil(() -> mgr.closed());
+
+    // Check state at the end
+    assertEquals(0, mgr.size());
+    assertEquals(0, mgr.pool.waitersCount());
+    assertEquals(0, mgr.pool.waitersInQueue());
+    assertEquals(0, mgr.pool.weight());
+    assertEquals(0, mgr.pool.capacity());
   }
 
   class FakeWaiter extends Waiter<FakeConnection> {