fix(qwp): I/O thread disposes its client on exit; a connect racing close() discards instead of installing (C5)

Root cause of the C5 leak: ownership hand-off between close() and an
in-flight connect attempt was broken. close() swallows the
InterruptedException from shutdownLatch.await() (routine under
BackgroundDrainerPool.close()'s shutdownNow escalation, now that
Invariant-B drainers retry outages indefinitely) and returns while the
I/O thread is still inside a blocking native connect (connect_timeout
defaults to 0 = OS timeout; neither unpark nor interrupt cancels
connect(2)). Its client read saw the pre-connect field; a subsequently
succeeding reconnect() then installed a live client via swapClient into
the abandoned loop, and no code path ever closed it -- native socket,
fds and buffers leaked for the life of the process.

Fix, in two layers, both on the I/O thread (close() stays
interrupt-responsive; the re-await alternative would block user close()
calls for up to the OS connect timeout):

1. connectLoop: a client obtained after running flipped false is closed
   and discarded instead of installed. This also stops the routine
   post-teardown engine access via swapClient -> positionCursorAt ->
   Unsafe reads against segments the drainer's finally may have already
   unmapped (the SEGV half of C5).
2. ioLoop's finally: the I/O thread disposes whatever client it holds
   before counting down the shutdown latch -- the airtight backstop for
   running flipping false between connectLoop's check and swapClient
   completing. Duplicate closes (loop.close()'s own, owners' stale
   references) remain no-ops: WebSocketClient.close() is idempotent.

Turns CursorWebSocketSendLoopInterruptedCloseLeakTest green. Regression
net: sf/cursor (309), qwp/client (895), impl + http/client (146) -- all
green.

Residual (out of scope here): the drainer-side teardown-overlap window
-- interrupt landing in the nanoseconds between the running check and
swapClient while the engine unmaps -- needs the ioThreadStopped-style
guard in BackgroundDrainer plus a close() that signals a failed stop;
tracked as the SEGV half of the C5 review finding.

Author: bluestreak01

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java

@@ -586,9 +586,11 @@ public synchronized void close() {
         // replaced the original (and closed it); the owner only retains
         // the stale pre-reconnect reference. Without closing the live
         // client here, its native socket and fds leak past sender.close()
-        // every time the loop reconnected at least once. close() is
-        // idempotent, so the owner's duplicate close on its stale
-        // reference is still safe.
+        // every time the loop reconnected at least once. ioLoop's finally
+        // also closes the current client on I/O-thread exit, so this read
+        // matters chiefly when the loop never started (SYNC construction,
+        // close() before start()) — and doubles as a safety net. close()
+        // is idempotent, so duplicate closes on any path are safe.
         WebSocketClient c = client;
         if (c != null) {
             try {
@@ -898,6 +900,27 @@ private void connectLoop(Throwable initial, String phase) {
             try {
                 WebSocketClient newClient = reconnectFactory.reconnect();
                 if (newClient != null) {
+                    if (!running) {
+                        // close() ran while this connect attempt was in
+                        // flight. Its latch await may have been interrupted
+                        // (BackgroundDrainerPool.close()'s shutdownNow path)
+                        // and returned already — the owner's teardown,
+                        // including the engine unmap in BackgroundDrainer's
+                        // finally, can be complete. Installing the client now
+                        // would (a) touch engine memory via positionCursorAt
+                        // after a possible unmap and (b) abandon a live socket
+                        // in a loop nothing will revisit — close() has run,
+                        // its client read saw the pre-connect field. The
+                        // attempt owns the client until it is installed, so
+                        // dispose of it here, on the I/O thread, and exit
+                        // through the quiet stopped path below.
+                        try {
+                            newClient.close();
+                        } catch (Throwable ignored) {
+                            // best-effort
+                        }
+                        break;
+                    }
                     swapClient(newClient);
                     totalReconnects.incrementAndGet();
                     long elapsedMs = (System.nanoTime() - outageStartNanos) / 1_000_000L;
@@ -1189,6 +1212,25 @@ private void ioLoop() {
             }
             fail(t);
         } finally {
+            // Last act of the I/O thread: dispose of whatever client it
+            // holds. This is the airtight half of the close()-vs-reconnect
+            // race — when close()'s latch await is interrupted (drainer pool
+            // shutdownNow), close() returns before this thread has exited,
+            // and its own client close saw the pre-reconnect field. A client
+            // swapped in by the tail of an in-flight connect attempt (running
+            // flipped false between connectLoop's check and swapClient) would
+            // be abandoned live without this. Runs BEFORE the latch countdown
+            // so a non-interrupted close() observes a fully disposed loop.
+            // Duplicate closes — loop.close()'s own, owners' stale references
+            // — stay safe: WebSocketClient.close() is idempotent.
+            WebSocketClient c = client;
+            if (c != null) {
+                try {
+                    c.close();
+                } catch (Throwable ignored) {
+                    // best-effort
+                }
+            }
             shutdownLatch.countDown();
         }
     }