Prevent use-after-free when IO thread doesn't stop

While closing the sender, we send the IO thread the stop signal and
wait for it to complete. If it doesn't complete on time, we can't
just continue with the teardown, freeing the resources it's still
using. We must bail out of teardown.

Author: mtopolnik

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -39,6 +39,7 @@
 import io.questdb.client.std.Decimal256;
 import io.questdb.client.std.Decimal64;
 import io.questdb.client.std.LongHashSet;
+import io.questdb.client.std.Misc;
 import io.questdb.client.std.ObjList;
 import io.questdb.client.std.bytes.DirectByteSlice;
 import org.jetbrains.annotations.NotNull;
@@ -403,6 +404,7 @@ public QwpWebSocketSender charColumn(CharSequence columnName, char value) {
     public void close() {
         if (!closed) {
             closed = true;
+            boolean ioThreadStopped = true;
 
             // Flush any remaining data
             try {
@@ -436,10 +438,16 @@ public void close() {
                 try {
                     sendQueue.close();
                 } catch (Exception e) {
+                    ioThreadStopped = false;
                     LOG.error("Error closing send queue: {}", String.valueOf(e));
                 }
             }
 
+            if (!ioThreadStopped) {
+                LOG.error("Skipping WebSocket client teardown because the I/O thread is still running");
+                return;
+            }
+
             // Close buffers (async mode only, window > 1)
             if (buffer0 != null) {
                 buffer0.close();
@@ -458,10 +466,7 @@ public void close() {
             for (int i = 0, n = keys.size(); i < n; i++) {
                 CharSequence key = keys.getQuick(i);
                 if (key != null) {
-                    QwpTableBuffer tb = tableBuffers.get(key);
-                    if (tb != null) {
-                        tb.close();
-                    }
+                    Misc.free(tableBuffers.get(key));
                 }
             }
             tableBuffers.clear();

core/src/main/java/io/questdb/client/cutlass/qwp/client/WebSocketSendQueue.java

@@ -205,11 +205,16 @@ public void close() {
         }
         ioThread.interrupt();
 
-        // Wait for I/O thread to finish
-        try {
-            shutdownLatch.await(shutdownTimeoutMs, TimeUnit.MILLISECONDS);
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
+        // Wait for I/O thread to finish before allowing the caller to free
+        // the socket and client-owned native buffers. If a send/recv call is
+        // still blocked, disconnect the socket to force it to unwind.
+        if (!awaitShutdown(shutdownTimeoutMs)) {
+            LOG.warn("I/O thread did not stop within {}ms, disconnecting socket", shutdownTimeoutMs);
+            client.disconnect();
+            ioThread.interrupt();
+            if (!awaitShutdown(shutdownTimeoutMs)) {
+                throw new LineSenderException("Timed out waiting for WebSocket I/O thread to stop");
+            }
         }
 
         LOG.info("WebSocket send queue closed [totalBatches={}, totalBytes={}]", totalBatchesSent.get(), totalBytesSent.get());
@@ -503,6 +508,15 @@ private boolean isPendingEmpty() {
         return pendingBuffer == null;
     }
 
+    private boolean awaitShutdown(long timeoutMs) {
+        try {
+            return shutdownLatch.await(timeoutMs, TimeUnit.MILLISECONDS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return shutdownLatch.getCount() == 0;
+        }
+    }
+
     private boolean offerPending(MicrobatchBuffer buffer) {
         if (pendingBuffer != null) {
             return false; // slot occupied