Prove sender-pool unwind in build-failure test

testQueryPoolBuildFailureUnwindsSenderPool previously asserted only that
build() threw when the query pool could not connect. That cannot catch
the bug it targets: if build() threw and also leaked the already-built
sender pool's connected senders, the test still passed.

TestWebSocketServer now tracks connections from the server's view. It
increments a live counter when a handshake completes and decrements it
when that connection's read thread exits (the client closed its socket),
and keeps a monotonic handshake counter. The read loop runs inside a
try/finally so the decrement fires on every exit path.

The test now proves the unwind two ways: the server saw two ingest
handshakes (the senders connected, so the next check is not vacuous),
and the live connection count returns to zero after the failed build
(the unwind closed every sender). The live count is polled because the
server observes the client-side close asynchronously.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: mtopolnik

core/src/test/java/io/questdb/client/test/QuestDBBuilderTest.java

@@ -31,6 +31,7 @@
 import org.junit.Test;
 
 import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
 
 public class QuestDBBuilderTest {
 
@@ -274,9 +275,21 @@ public void testQueryPoolBuildFailureUnwindsSenderPool() throws Exception {
                         .close();
                 Assert.fail("expected build to fail when query pool cannot connect");
             } catch (RuntimeException expected) {
-                // The exact exception comes from QwpQueryClient.connect(); the
-                // build failing tells us the sender-pool unwind ran.
+                // The exact exception comes from QwpQueryClient.connect(). The
+                // build failing only proves the query pool gave up; the
+                // assertions below prove the unwind closed the senders the
+                // sender pool had already connected, rather than leaking them.
             }
+            // The sender pool eagerly warmed senderPoolSize(2), so the server
+            // saw two ingest handshakes (proving the senders connected and the
+            // assertion below is not vacuous)...
+            awaitTrue("sender pool should have connected two ingest senders",
+                    () -> ingest.handshakeCount() >= 2);
+            // ...and the failed build() must have closed every one of them, so
+            // no sender connection is left live on the server. The server
+            // observes the client-side socket close asynchronously, so poll.
+            awaitTrue("failed build() must close the already-built sender pool, leaving no live connection",
+                    () -> ingest.liveConnectionCount() == 0);
         }
     }
 
@@ -396,4 +409,15 @@ private static void assertSchemaRejected(Runnable action) {
             Assert.assertTrue(e.getMessage(), e.getMessage().contains("ws or wss"));
         }
     }
+
+    private static void awaitTrue(String message, BooleanSupplier condition) throws InterruptedException {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+        while (System.nanoTime() < deadline) {
+            if (condition.getAsBoolean()) {
+                return;
+            }
+            Thread.sleep(20);
+        }
+        Assert.assertTrue(message, condition.getAsBoolean());
+    }
 }

core/src/test/java/io/questdb/client/test/cutlass/qwp/websocket/TestWebSocketServer.java

@@ -46,6 +46,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * A simple WebSocket server for client integration testing.
@@ -57,10 +58,19 @@ public class TestWebSocketServer implements Closeable {
     private final List<ClientHandler> clients = new CopyOnWriteArrayList<>();
     private final boolean emitDurableAckHeader;
     private final WebSocketServerHandler handler;
+    // Count of WebSocket connections currently live from the server's view:
+    // incremented when a handshake completes, decremented when that connection's
+    // read thread exits (the client closed its socket). Lets a test assert that a
+    // client-side pool actually closed the connections it opened.
+    private final AtomicInteger liveConnections = new AtomicInteger();
     private final int port;
     private final AtomicBoolean running = new AtomicBoolean(false);
     private final ServerSocket serverSocket;
     private final CountDownLatch startLatch = new CountDownLatch(1);
+    // Monotonic count of completed handshakes over the server's lifetime. Unlike
+    // liveConnections it never decrements, so a test can confirm how many clients
+    // connected even after they have all disconnected.
+    private final AtomicInteger totalHandshakes = new AtomicInteger();
     private Thread acceptThread;
     // X-QuestDB-Role value to emit on handshake responses. null = omit the
     // header (legacy behavior for tests written before role-aware failover).
@@ -164,6 +174,22 @@ public int getPort() {
         return port;
     }
 
+    /**
+     * Number of handshakes the server has completed over its lifetime
+     * (monotonic; never decreases when clients disconnect).
+     */
+    public int handshakeCount() {
+        return totalHandshakes.get();
+    }
+
+    /**
+     * Number of WebSocket connections currently live from the server's view.
+     * Drops back to zero once every client has closed its socket.
+     */
+    public int liveConnectionCount() {
+        return liveConnections.get();
+    }
+
     /**
      * Replaces the advertised role for subsequent handshakes (live update).
      */
@@ -536,35 +562,41 @@ void start() {
                         LOG.error("Handshake failed");
                         return;
                     }
+                    totalHandshakes.incrementAndGet();
+                    liveConnections.incrementAndGet();
 
-                    if (sendServerInfo) {
-                        sendBinary(buildServerInfoFrame(roleByte(advertisedRole)));
-                    }
-
-                    byte[] readBuf = new byte[8192];
-
-                    while (running.get() && !isClosed) {
-                        int read;
-                        try {
-                            read = in.read(readBuf);
-                        } catch (SocketTimeoutException e) {
-                            continue;
-                        }
-                        if (read <= 0) {
-                            break;
+                    try {
+                        if (sendServerInfo) {
+                            sendBinary(buildServerInfoFrame(roleByte(advertisedRole)));
                         }
 
-                        // append to recvBuffer
-                        recvBuffer.compact();
-                        if (recvBuffer.remaining() < read) {
-                            // should not happen with 64k buffer in tests
-                            LOG.error("Receive buffer overflow");
-                            break;
+                        byte[] readBuf = new byte[8192];
+
+                        while (running.get() && !isClosed) {
+                            int read;
+                            try {
+                                read = in.read(readBuf);
+                            } catch (SocketTimeoutException e) {
+                                continue;
+                            }
+                            if (read <= 0) {
+                                break;
+                            }
+
+                            // append to recvBuffer
+                            recvBuffer.compact();
+                            if (recvBuffer.remaining() < read) {
+                                // should not happen with 64k buffer in tests
+                                LOG.error("Receive buffer overflow");
+                                break;
+                            }
+                            recvBuffer.put(readBuf, 0, read);
+                            recvBuffer.flip();
+
+                            handleRead();
                         }
-                        recvBuffer.put(readBuf, 0, read);
-                        recvBuffer.flip();
-
-                        handleRead();
+                    } finally {
+                        liveConnections.decrementAndGet();
                     }
                 } catch (IOException e) {
                     if (running.get()) {