fix(qwp): finite default connect deadline for background drainers; CAS-guard WebSocketClient.close() (C4 residuals)

With connect_timeout unset (0), a drainer parked in an untimed native
connect (nf.connectAddrInfo, OS-bounded: 60-130s of SYN retries on Linux)
is the normal state during an outage. The pool's shutdownNow path (~3s
into sender.close()) then reliably lands on the failed-stop teardown
protocol: the WebSocket client and microbatch buffers are deliberately
leaked and the SF slot lock is held until the OS deadline resolves the
stuck connect. Background connect walks now default to a finite 15s
deadline; an explicit connect_timeout is honoured verbatim on both paths
and the foreground's unset default stays untimed (no behavior change).

WebSocketClient.close() frees native memory but was gated by a volatile
check-then-act -- two concurrent closers could both enter and double-run
disconnect()/Unsafe.free. Now a CAS on an AtomicBoolean: exactly one
closer runs teardown. No concurrent-closer path is reachable at HEAD
(the failed-stop protocol severed them), so this is defense in depth
that makes the 'duplicate closes are safe' comments in
CursorWebSocketSendLoop true unconditionally rather than
by-current-call-graph.

Tests: BackgroundConnectTimeoutDefaultTest covers the resolver matrix;
WebSocketClientTest.testConcurrentCloseRunsTeardownExactlyOnce races
4 closers x 200 iterations under assertMemoryLeak, where a double-free
surfaces as a native memory-counter mismatch.

Author: bluestreak01

core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java

@@ -47,6 +47,7 @@
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.util.Base64;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
 
@@ -99,8 +100,11 @@ public abstract class WebSocketClient implements QuietCloseable {
     private final int maxRecvBufSize;
     private final SecureRnd rnd;
     private final WebSocketSendBuffer sendBuffer;
-    // volatile: written by user thread in close(), read by I/O thread in checkConnected()/sendFrame()/receiveFrame()
-    private volatile boolean closed;
+    // Written by whichever closer wins the CAS in close(); read by the I/O
+    // thread in checkConnected()/sendFrame()/receiveFrame(). An AtomicBoolean
+    // (not a bare volatile check-then-act) so concurrent closers cannot both
+    // enter close() and double-run disconnect()/Unsafe.free.
+    private final AtomicBoolean closed = new AtomicBoolean();
     // Upper bound (ms) on the TCP connect. <= 0 disables the application-level
     // timeout and falls back to the OS connect timeout. Seeded from the
     // configuration; the QWP sender may override it via setConnectTimeout().
@@ -197,7 +201,7 @@ public WebSocketClient(HttpClientConfiguration configuration, SocketFactory sock
             this.frameParser = new WebSocketFrameParser();
             this.rnd = new SecureRnd();
             this.upgraded = false;
-            this.closed = false;
+            this.closed.set(false);
         } catch (Throwable t) {
             if (recvBufPtr != 0) {
                 Unsafe.free(recvBufPtr, recvBufSize, MemoryTag.NATIVE_DEFAULT);
@@ -212,8 +216,12 @@ public WebSocketClient(HttpClientConfiguration configuration, SocketFactory sock
 
     @Override
     public void close() {
-        if (!closed) {
-            closed = true;
+        // CAS gate: exactly one closer runs the teardown below. Closers can be
+        // the owner thread, the I/O thread's exit path, or a stale duplicate
+        // reference (see CursorWebSocketSendLoop) -- a bare volatile
+        // check-then-act here would let two concurrent closers both enter and
+        // double-run disconnect()/Unsafe.free (native double-free).
+        if (closed.compareAndSet(false, true)) {
 
             // Try to send close frame
             if (upgraded && !socket.isClosed()) {
@@ -247,7 +255,7 @@ public void close() {
      * @param port the server port
      */
     public void connect(CharSequence host, int port) {
-        if (closed) {
+        if (closed.get()) {
             throw new HttpClientException("WebSocket client is closed");
         }
 
@@ -380,7 +388,7 @@ public int getUpgradeStatusCode() {
      * Returns whether the WebSocket is connected and upgraded.
      */
     public boolean isConnected() {
-        return upgraded && !closed && !socket.isClosed();
+        return upgraded && !closed.get() && !socket.isClosed();
     }
 
     /**
@@ -585,7 +593,7 @@ public boolean tryReceiveFrame(WebSocketFrameHandler handler) {
      * @param authorizationHeader the Authorization header value (e.g., "Basic ..."), or null
      */
     public void upgrade(CharSequence path, int timeout, CharSequence authorizationHeader) {
-        if (closed) {
+        if (closed.get()) {
             throw new HttpClientException("WebSocket client is closed");
         }
         if (socket.isClosed()) {
@@ -892,7 +900,7 @@ private void appendToFragmentBuffer(long payloadPtr, int payloadLen) {
     }
 
     private void checkConnected() {
-        if (closed) {
+        if (closed.get()) {
             throw new HttpClientException("WebSocket client is closed");
         }
         if (!upgraded) {

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

@@ -127,6 +127,9 @@ public class QwpWebSocketSender implements Sender {
     public static final int DEFAULT_AUTO_FLUSH_BYTES = 8 * 1024 * 1024;
     public static final long DEFAULT_AUTO_FLUSH_INTERVAL_NANOS = 100_000_000L; // 100ms
     public static final int DEFAULT_AUTO_FLUSH_ROWS = 1_000;
+    // Finite fallback (ms) for BACKGROUND (drainer) TCP connects when the
+    // user left connect_timeout unset. See effectiveConnectTimeoutMs.
+    public static final int DEFAULT_BACKGROUND_CONNECT_TIMEOUT_MS = 15_000;
     private static final int DEFAULT_BUFFER_SIZE = 8192;
     private static final int DEFAULT_MICROBATCH_BUFFER_SIZE = 1024 * 1024; // 1MB
     private static final Logger LOG = LoggerFactory.getLogger(QwpWebSocketSender.class);
@@ -2445,6 +2448,25 @@ private void atNanos(long timestampNanos) {
         sendRow();
     }
 
+    /**
+     * Resolves the connect timeout for one {@code buildAndConnect} walk.
+     * Foreground connects honour the configured value verbatim: 0 (the
+     * default) keeps the historical untimed native connect, bounded only by
+     * the OS (SYN retries, 60-130s on Linux). Background (drainer) connects
+     * get a finite fallback instead: during an outage a drainer is routinely
+     * parked inside a blocking native connect that neither unpark nor
+     * interrupt cancels, so the drainer pool's shutdownNow path (~3s into
+     * sender.close()) reliably lands on the failed-stop protocol -- the
+     * WebSocket client and microbatch buffers are deliberately leaked and
+     * the slot lock is held until the OS deadline resolves the connect. A
+     * finite background deadline bounds that window to seconds without
+     * changing foreground semantics. Exposed for unit tests.
+     */
+    @TestOnly
+    public static int effectiveConnectTimeoutMs(boolean background, int configuredMs) {
+        return background && configuredMs <= 0 ? DEFAULT_BACKGROUND_CONNECT_TIMEOUT_MS : configuredMs;
+    }
+
     private synchronized WebSocketClient buildAndConnect(ReconnectSupplier ctx) {
         // Background (drainer) factories share this connect walk -- endpoint
         // list, hostTracker health and round state -- but must stay INVISIBLE
@@ -2517,7 +2539,7 @@ private synchronized WebSocketClient buildAndConnect(ReconnectSupplier ctx) {
                 newClient.setQwpMaxVersion(QwpConstants.VERSION);
                 newClient.setQwpClientId(QwpConstants.CLIENT_ID);
                 newClient.setQwpRequestDurableAck(requestDurableAck);
-                newClient.setConnectTimeout(connectTimeoutMs);
+                newClient.setConnectTimeout(effectiveConnectTimeoutMs(background, connectTimeoutMs));
                 newClient.connect(ep.host, ep.port);
                 int upgradeTimeoutMs = (int) Math.min(authTimeoutMs, Integer.MAX_VALUE);
                 newClient.upgrade(WRITE_PATH, upgradeTimeoutMs, authorizationHeader);

core/src/test/java/io/questdb/client/test/cutlass/http/client/WebSocketClientTest.java

@@ -37,11 +37,55 @@
 
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static io.questdb.client.test.tools.TestUtils.assertMemoryLeak;
 
 public class WebSocketClientTest {
 
+    /**
+     * close() frees native memory (recv/fragment buffers, send buffers), so
+     * its guard must be a CAS, not a volatile check-then-act: two concurrent
+     * closers passing the flag check together would both run
+     * disconnect()/Unsafe.free -- a native double-free. Closers can race in
+     * practice: the owner thread's teardown vs the I/O thread's exit path vs
+     * stale duplicate references (see CursorWebSocketSendLoop). The memory
+     * counters checked by assertMemoryLeak flag a double-free as a counter
+     * mismatch.
+     */
+    @Test
+    public void testConcurrentCloseRunsTeardownExactlyOnce() throws Exception {
+        assertMemoryLeak(() -> {
+            final int threads = 4;
+            final int iterations = 200;
+            for (int i = 0; i < iterations; i++) {
+                StubWebSocketClient client = new StubWebSocketClient();
+                CyclicBarrier barrier = new CyclicBarrier(threads);
+                AtomicReference<Throwable> failure = new AtomicReference<>();
+                Thread[] closers = new Thread[threads];
+                for (int t = 0; t < threads; t++) {
+                    closers[t] = new Thread(() -> {
+                        try {
+                            barrier.await();
+                            client.close();
+                        } catch (Throwable e) {
+                            failure.compareAndSet(null, e);
+                        }
+                    });
+                    closers[t].start();
+                }
+                for (Thread closer : closers) {
+                    closer.join();
+                }
+                Throwable t = failure.get();
+                if (t != null) {
+                    throw new AssertionError("concurrent close failed on iteration " + i, t);
+                }
+            }
+        });
+    }
+
     @Test
     public void testExtractMaxBatchSizeAbsentHeaderReturnsZero() throws Exception {
         String response = "HTTP/1.1 101 Switching Protocols\r\n"

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/BackgroundConnectTimeoutDefaultTest.java

@@ -0,0 +1,81 @@
+/*+*****************************************************************************
+ *     ___                  _   ____  ____
+ *    / _ \ _   _  ___  ___| |_|  _ \| __ )
+ *   | | | | | | |/ _ \/ __| __| | | |  _ \
+ *   | |_| | |_| |  __/\__ \ |_| |_| | |_) |
+ *    \__\_\\__,_|\___||___/\__|____/|____/
+ *
+ *  Copyright (c) 2014-2019 Appsicle
+ *  Copyright (c) 2019-2026 QuestDB
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *
+ ******************************************************************************/
+
+package io.questdb.client.test.cutlass.qwp.client;
+
+import io.questdb.client.cutlass.qwp.client.QwpWebSocketSender;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static io.questdb.client.cutlass.qwp.client.QwpWebSocketSender.DEFAULT_BACKGROUND_CONNECT_TIMEOUT_MS;
+import static io.questdb.client.cutlass.qwp.client.QwpWebSocketSender.effectiveConnectTimeoutMs;
+
+/**
+ * Background (drainer) connect walks must never inherit the untimed native
+ * connect that connect_timeout=0 (the default) means for the foreground.
+ * <p>
+ * During an outage a drainer is routinely parked inside a blocking native
+ * connect ({@code nf.connectAddrInfo}) that neither unpark nor interrupt
+ * cancels. The drainer pool's close sequence (2.5s graceful drain +
+ * requestStop + 500ms + shutdownNow) then reliably lands on the failed-stop
+ * teardown protocol: the WebSocket client and microbatch buffers are
+ * deliberately leaked and the SF slot lock is held until the OS connect
+ * deadline (SYN retries, 60-130s on Linux) resolves the stuck call. A finite
+ * background default bounds that window to seconds. Foreground semantics are
+ * intentionally untouched: an explicit user value is honoured verbatim on
+ * both paths, and the foreground's unset default stays untimed.
+ */
+public class BackgroundConnectTimeoutDefaultTest {
+
+    @Test
+    public void testBackgroundExplicitValueHonoured() {
+        Assert.assertEquals(500, effectiveConnectTimeoutMs(true, 500));
+        Assert.assertEquals(60_000, effectiveConnectTimeoutMs(true, 60_000));
+    }
+
+    @Test
+    public void testBackgroundUnsetGetsFiniteDefault() {
+        Assert.assertEquals(DEFAULT_BACKGROUND_CONNECT_TIMEOUT_MS, effectiveConnectTimeoutMs(true, 0));
+        // Defensive: builder validation rejects negatives, but the resolver
+        // must not turn a bad value back into an untimed background connect.
+        Assert.assertEquals(DEFAULT_BACKGROUND_CONNECT_TIMEOUT_MS, effectiveConnectTimeoutMs(true, -1));
+    }
+
+    @Test
+    public void testDefaultIsFinite() {
+        Assert.assertTrue(DEFAULT_BACKGROUND_CONNECT_TIMEOUT_MS > 0);
+    }
+
+    @Test
+    public void testForegroundExplicitValueHonoured() {
+        Assert.assertEquals(500, effectiveConnectTimeoutMs(false, 500));
+    }
+
+    @Test
+    public void testForegroundUnsetStaysUntimed() {
+        // 0 => WebSocketClient falls back to nf.connectAddrInfo (OS-bounded).
+        // Historical foreground behaviour, deliberately preserved.
+        Assert.assertEquals(0, effectiveConnectTimeoutMs(false, 0));
+    }
+}