feat(net): add application-level TCP connect timeout

Establish a real, cross-platform connect timeout for the HTTP and
WebSocket (QWP) transports. Previously a connect to a black-holed or
firewalled host blocked on the OS-level TCP connect timeout (often
60-120s) because the socket was created blocking and only switched to
non-blocking *after* connect; the transport exposed no knob to clamp it.

Approach: a new native primitive switches the socket to non-blocking
*before* connect, so connect() returns EINPROGRESS immediately, then
polls for writability bounded by the caller's budget and confirms the
outcome via SO_ERROR. A distinct return code (CONNECT_TIMEOUT, -3) lets
the Java layer raise a timeout-flagged exception rather than decode errno.

Native:
- share/net.c: connectAddrInfoTimeout + awaitConnectComplete (poll +
  getsockopt(SO_ERROR), monotonic-clock EINTR handling)
- windows/net.c: Winsock equivalent (select write/except sets)
- share/net.h: ECONNTIMEOUT (-3) sentinel

Java:
- Net / NetworkFacade(Impl): connectAddrInfoTimeout + CONNECT_TIMEOUT
- HttpClientConfiguration.getConnectTimeout() (default 0 = OS fallback)
- HttpClient.connect() / WebSocketClient.doConnect() honor it and throw a
  timeout-flagged HttpClientException on CONNECT_TIMEOUT
- Sender builder: connectTimeoutMillis() + connect_timeout connect-string
  key (legacy http and ws/wss parsers) + ConfigSchema COMMON key
- QwpWebSocketSender / QwpQueryClient: thread the value through to their
  WebSocketClient (adds QwpQueryClient.withConnectTimeout)

Default is unset (0): behaviour is unchanged unless connect_timeout is
configured.

Tests: NetConnectTimeoutTest covers loopback success, refused-vs-timeout
disambiguation, and a black-hole timeout that fires within budget;
config-honored drift guards updated for the new COMMON key.

Author: bluestreak01

core/src/main/c/share/net.c

@@ -33,6 +33,8 @@
 #include <stdlib.h>
 #include <stdint.h>
 #include <string.h>
+#include <poll.h>
+#include <time.h>
 #include "net.h"
 #include <netdb.h>
 #include "sysutil.h"
@@ -298,6 +300,85 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_connectAddrInfo
     return handleEintrInConnect(fd, result);
 }
 
+// Waits up to timeout_millis for an in-progress non-blocking connect on fd to
+// finish. Returns 0 on success, -1 on connection failure (errno set so the
+// caller can read it via Os.errno()), or com_questdb_network_Net_ECONNTIMEOUT
+// on timeout.
+static jint awaitConnectComplete(int fd, jint timeout_millis) {
+    struct timespec start;
+    clock_gettime(CLOCK_MONOTONIC, &start);
+
+    for (;;) {
+        struct pollfd pfd;
+        pfd.fd = fd;
+        pfd.events = POLLOUT;
+        pfd.revents = 0;
+
+        int wait_millis = timeout_millis > 0 ? timeout_millis : 0;
+        int rc = poll(&pfd, 1, wait_millis);
+        if (rc > 0) {
+            // The connect attempt has finished one way or another; the only
+            // authoritative result is SO_ERROR (POLLOUT alone does not mean
+            // success -- a refused connection is also reported as writable).
+            int so_error = 0;
+            socklen_t len = sizeof(so_error);
+            if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &so_error, &len) < 0) {
+                return -1;
+            }
+            if (so_error != 0) {
+                errno = so_error;
+                return -1;
+            }
+            return 0;
+        }
+        if (rc == 0) {
+            errno = ETIMEDOUT;
+            return com_questdb_network_Net_ECONNTIMEOUT;
+        }
+        if (errno != EINTR) {
+            return -1;
+        }
+        // Interrupted by a signal: recompute the remaining budget against a
+        // monotonic clock so EINTR storms cannot extend the timeout, and retry.
+        struct timespec now;
+        clock_gettime(CLOCK_MONOTONIC, &now);
+        long elapsed = (now.tv_sec - start.tv_sec) * 1000L
+                       + (now.tv_nsec - start.tv_nsec) / 1000000L;
+        start = now;
+        timeout_millis = (jint) (timeout_millis - elapsed);
+        if (timeout_millis <= 0) {
+            errno = ETIMEDOUT;
+            return com_questdb_network_Net_ECONNTIMEOUT;
+        }
+    }
+}
+
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_connectAddrInfoTimeout
+        (JNIEnv *e, jclass cl, jint fd, jlong lpAddrInfo, jint timeoutMillis) {
+    struct addrinfo *addr = (struct addrinfo *) lpAddrInfo;
+
+    // Switch to non-blocking BEFORE connect so connect() returns immediately
+    // with EINPROGRESS instead of blocking on the OS connect timeout. The
+    // socket is left non-blocking on success, matching the post-connect
+    // configureNonBlocking() the callers already perform.
+    int flags = fcntl((int) fd, F_GETFL, 0);
+    if (flags < 0) {
+        return -1;
+    }
+    if (fcntl((int) fd, F_SETFL, flags | O_NONBLOCK) < 0) {
+        return -1;
+    }
+
+    int result = connect((int) fd, addr->ai_addr, (int) addr->ai_addrlen);
+    if (result == 0) {
+        return 0; // connected immediately (e.g. loopback)
+    }
+    if (errno == EINPROGRESS || errno == EINTR || errno == EWOULDBLOCK) {
+        return awaitConnectComplete((int) fd, timeoutMillis);
+    }
+    return -1; // immediate failure, errno set
+}
+
 JNIEXPORT void JNICALL Java_io_questdb_client_network_Net_freeAddrInfo0
         (JNIEnv *e, jclass cl, jlong address) {
     if (address != 0) {

core/src/main/c/share/net.h

@@ -160,6 +160,66 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_connectAddrInfo
     return res;
 }
 
+JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_connectAddrInfoTimeout
+        (JNIEnv *e, jclass cl, jint fd, jlong lpAddrInfo, jint timeoutMillis) {
+    struct addrinfo *addr = (struct addrinfo *) lpAddrInfo;
+    SOCKET s = (SOCKET) fd;
+
+    // Switch to non-blocking BEFORE connect so it returns immediately with
+    // WSAEWOULDBLOCK instead of blocking on the OS connect timeout.
+    u_long mode = 1;
+    if (ioctlsocket(s, FIONBIO, &mode) != 0) {
+        SaveLastError();
+        return -1;
+    }
+
+    int res = connect(s, addr->ai_addr, (int) addr->ai_addrlen);
+    if (res == 0) {
+        return 0; // connected immediately (e.g. loopback)
+    }
+    if (WSAGetLastError() != WSAEWOULDBLOCK) {
+        SaveLastError();
+        return -1;
+    }
+
+    fd_set writefds, exceptfds;
+    FD_ZERO(&writefds);
+    FD_ZERO(&exceptfds);
+    FD_SET(s, &writefds);
+    FD_SET(s, &exceptfds);
+
+    struct timeval tv;
+    tv.tv_sec = timeoutMillis / 1000;
+    tv.tv_usec = (timeoutMillis % 1000) * 1000;
+
+    // Winsock signals a failed non-blocking connect via the exception set.
+    int sel = select(0, NULL, &writefds, &exceptfds, &tv);
+    if (sel == 0) {
+        WSASetLastError(WSAETIMEDOUT);
+        SaveLastError();
+        return com_questdb_network_Net_ECONNTIMEOUT;
+    }
+    if (sel == SOCKET_ERROR) {
+        SaveLastError();
+        return -1;
+    }
+
+    int so_error = 0;
+    int len = sizeof(so_error);
+    if (FD_ISSET(s, &exceptfds) || !FD_ISSET(s, &writefds)) {
+        getsockopt(s, SOL_SOCKET, SO_ERROR, (char *) &so_error, &len);
+        WSASetLastError(so_error != 0 ? so_error : WSAECONNREFUSED);
+        SaveLastError();
+        return -1;
+    }
+    if (getsockopt(s, SOL_SOCKET, SO_ERROR, (char *) &so_error, &len) == 0 && so_error != 0) {
+        WSASetLastError(so_error);
+        SaveLastError();
+        return -1;
+    }
+    return 0;
+}
+
 JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_configureNonBlocking
         (JNIEnv *e, jclass cl, jint fd) {
     u_long mode = 1;

core/src/main/c/windows/net.c

@@ -38,6 +38,15 @@ default boolean fixBrokenConnection() {
         return true;
     }
 
+    /**
+     * Upper bound, in milliseconds, on establishing the TCP connection. When
+     * {@code <= 0} (the default) no application-level connect timeout is applied
+     * and the connect falls back to the OS-level TCP connect timeout.
+     */
+    default int getConnectTimeout() {
+        return 0;
+    }
+
     default EpollFacade getEpollFacade() {
         return EpollFacadeImpl.INSTANCE;
     }

core/src/main/java/io/questdb/client/HttpClientConfiguration.java

@@ -1011,6 +1011,9 @@ final class LineSenderBuilder {
         private int autoFlushRows = PARAMETER_NOT_SET_EXPLICITLY;
         private int bufferCapacity = PARAMETER_NOT_SET_EXPLICITLY;
         private long closeFlushTimeoutMillis = CLOSE_FLUSH_TIMEOUT_NOT_SET;
+        // Upper bound (ms) on the TCP connect. PARAMETER_NOT_SET_EXPLICITLY ->
+        // 0 (no application-level connect timeout; OS connect timeout applies).
+        private int connectTimeoutMillis = PARAMETER_NOT_SET_EXPLICITLY;
         // Optional user-supplied async connection-event listener. When null,
         // the sender uses DefaultSenderConnectionListener.INSTANCE
         // (loud-not-silent log of every transition).
@@ -1078,6 +1081,11 @@ public String getSettingsPath() {
             public int getTimeout() {
                 return httpTimeout == PARAMETER_NOT_SET_EXPLICITLY ? DEFAULT_HTTP_TIMEOUT : httpTimeout;
             }
+
+            @Override
+            public int getConnectTimeout() {
+                return connectTimeoutMillis == PARAMETER_NOT_SET_EXPLICITLY ? 0 : connectTimeoutMillis;
+            }
         };
         private long minRequestThroughput = PARAMETER_NOT_SET_EXPLICITLY;
         private int multicastTtl = PARAMETER_NOT_SET_EXPLICITLY;
@@ -1199,6 +1207,28 @@ public AdvancedTlsSettings advancedTls() {
             return new AdvancedTlsSettings();
         }
 
+        /**
+         * Upper bound, in milliseconds, on establishing the TCP connection to a
+         * QuestDB endpoint. When set, a connect that does not complete within
+         * this budget is aborted (instead of riding the much longer OS-level
+         * connect timeout). Applies to both HTTP/WebSocket transports. Default
+         * is unset (0), which falls back to the OS connect timeout.
+         *
+         * @param millis connect timeout in milliseconds; must be > 0
+         * @return this instance for method chaining
+         */
+        public LineSenderBuilder connectTimeoutMillis(int millis) {
+            if (this.connectTimeoutMillis != PARAMETER_NOT_SET_EXPLICITLY) {
+                throw new LineSenderException("connect timeout was already configured ")
+                        .put("[connect_timeout=").put(this.connectTimeoutMillis).put("]");
+            }
+            if (millis <= 0) {
+                throw new LineSenderException("connect_timeout must be > 0: ").put(millis);
+            }
+            this.connectTimeoutMillis = millis;
+            return this;
+        }
+
         /**
          * Per-endpoint timeout on the WebSocket upgrade response read. Default
          * {@value QwpWebSocketSender#DEFAULT_AUTH_TIMEOUT_MS} ms.
@@ -1531,6 +1561,7 @@ public Sender build() {
                             actualErrorInboxCapacity,
                             actualDurableAckKeepaliveIntervalMillis,
                             authTimeoutMillis,
+                            connectTimeoutMillis == PARAMETER_NOT_SET_EXPLICITLY ? 0 : connectTimeoutMillis,
                             connectionListener,
                             actualConnectionListenerInboxCapacity
                     );
@@ -3166,6 +3197,9 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                     pos = getValue(configurationString, pos, sink, "request_timeout");
                     int requestTimeout = parseIntValue(sink, "request_timeout");
                     httpTimeoutMillis(requestTimeout);
+                } else if (Chars.equals("connect_timeout", sink)) {
+                    pos = getValue(configurationString, pos, sink, "connect_timeout");
+                    connectTimeoutMillis(parseIntValue(sink, "connect_timeout"));
                 } else if (Chars.equals("request_min_throughput", sink)) {
                     pos = getValue(configurationString, pos, sink, "request_min_throughput");
                     int requestMinThroughput = parseIntValue(sink, "request_min_throughput");
@@ -3446,6 +3480,9 @@ private LineSenderBuilder fromConfigWebSocket(CharSequence configurationString)
                 if (view.has("auth_timeout_ms")) {
                     authTimeoutMillis(view.getLong("auth_timeout_ms", 0));
                 }
+                if (view.has("connect_timeout")) {
+                    connectTimeoutMillis((int) view.getLong("connect_timeout", 0));
+                }
 
                 s = view.getStr("auto_flush_rows");
                 if (s != null) {
@@ -3701,6 +3738,7 @@ public java.util.Map<String, Object> wsConfigSnapshotForTest() {
             m.put("connection_listener_inbox_capacity", connectionListenerInboxCapacity);
             m.put("token", httpToken);
             m.put("auth_timeout_ms", authTimeoutMillis);
+            m.put("connect_timeout", connectTimeoutMillis == PARAMETER_NOT_SET_EXPLICITLY ? 0 : connectTimeoutMillis);
             m.put("username", username);
             m.put("password", password);
             m.put("tls_verify", tlsValidationMode == null ? null : tlsValidationMode.name());

core/src/main/java/io/questdb/client/Sender.java

@@ -66,6 +66,7 @@ public abstract class HttpClient implements QuietCloseable {
     protected final NetworkFacade nf;
     protected final Socket socket;
     private final ObjectPool<DirectUtf8String> csPool = new ObjectPool<>(DirectUtf8String.FACTORY, 64);
+    private final int connectTimeout;
     private final int defaultTimeout;
     private final boolean fixBrokenConnection;
     private final int maxBufferSize;
@@ -84,6 +85,7 @@ public HttpClient(HttpClientConfiguration configuration, SocketFactory socketFac
         this.nf = configuration.getNetworkFacade();
         this.socket = socketFactory.newInstance(nf, LOG);
         this.defaultTimeout = configuration.getTimeout();
+        this.connectTimeout = configuration.getConnectTimeout();
         this.bufferSize = configuration.getInitialRequestBufferSize();
         this.maxBufferSize = configuration.getMaximumRequestBufferSize();
         this.responseParserBufSize = configuration.getResponseBufferSize();
@@ -617,10 +619,16 @@ private void connect(CharSequence host, int port) {
                 throw new HttpClientException("could not resolve host ").put("[host=").put(host).put("]");
             }
 
-            if (nf.connectAddrInfo(fd, addrInfo) != 0) {
+            final int connectResult = connectTimeout > 0
+                    ? nf.connectAddrInfoTimeout(fd, addrInfo, connectTimeout)
+                    : nf.connectAddrInfo(fd, addrInfo);
+            if (connectResult != 0) {
                 int errno = nf.errno();
                 nf.freeAddrInfo(addrInfo);
                 disconnect();
+                if (connectResult == NetworkFacade.CONNECT_TIMEOUT) {
+                    throw new HttpClientException("connect timed out ").put("[host=").put(host).put(", port=").put(port).put(", timeout=").put(connectTimeout).put(']').flagAsTimeout();
+                }
                 throw new HttpClientException("could not connect to host ").put("[host=").put(host).put(", port=").put(port).put(", errno=").put(errno).put(']');
             }
             nf.freeAddrInfo(addrInfo);

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

@@ -101,6 +101,10 @@ public abstract class WebSocketClient implements QuietCloseable {
     private final WebSocketSendBuffer sendBuffer;
     // volatile: written by user thread in close(), read by I/O thread in checkConnected()/sendFrame()/receiveFrame()
     private volatile boolean closed;
+    // 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().
+    private int connectTimeoutMillis;
     private int fragmentBufPos;
     private long fragmentBufPtr;       // native buffer for accumulating fragment payloads
     private int fragmentBufSize;
@@ -168,6 +172,7 @@ public WebSocketClient(HttpClientConfiguration configuration, SocketFactory sock
         this.nf = configuration.getNetworkFacade();
         this.socket = socketFactory.newInstance(nf, LOG);
         this.defaultTimeout = configuration.getTimeout();
+        this.connectTimeoutMillis = configuration.getConnectTimeout();
 
         int sendBufSize = Math.max(configuration.getInitialRequestBufferSize(), DEFAULT_SEND_BUFFER_SIZE);
         int maxSendBufSize = Math.max(configuration.getMaximumRequestBufferSize(), sendBufSize);
@@ -481,6 +486,16 @@ public void sendPing(int timeout) {
         }
     }
 
+    /**
+     * Overrides the TCP connect timeout (milliseconds) for subsequent
+     * {@link #connect} calls. {@code <= 0} disables the application-level
+     * timeout and falls back to the OS connect timeout. Must be called before
+     * {@link #connect}.
+     */
+    public void setConnectTimeout(int connectTimeoutMillis) {
+        this.connectTimeoutMillis = connectTimeoutMillis;
+    }
+
     /**
      * Sets the value sent as the {@code X-QWP-Accept-Encoding} upgrade header,
      * e.g. {@code "zstd;level=1,raw"}. Pass {@code null} to omit the header
@@ -922,10 +937,18 @@ private void doConnect(CharSequence host, int port) {
             throw new HttpClientException("could not resolve host [host=").put(host).put(']');
         }
 
-        if (nf.connectAddrInfo(fd, addrInfo) != 0) {
+        final int connectResult = connectTimeoutMillis > 0
+                ? nf.connectAddrInfoTimeout(fd, addrInfo, connectTimeoutMillis)
+                : nf.connectAddrInfo(fd, addrInfo);
+        if (connectResult != 0) {
             int errno = nf.errno();
             nf.freeAddrInfo(addrInfo);
             disconnect();
+            if (connectResult == NetworkFacade.CONNECT_TIMEOUT) {
+                throw new HttpClientException("connect timed out [host=").put(host)
+                        .put(", port=").put(port)
+                        .put(", timeout=").put(connectTimeoutMillis).put(']').flagAsTimeout();
+            }
             throw new HttpClientException("could not connect [host=").put(host)
                     .put(", port=").put(port)
                     .put(", errno=").put(errno).put(']');