fix(ilp): fix a leaked socket and native memory when an HTTP sender fails to start

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

@@ -46,7 +46,10 @@
 public class HttpHeaderParser implements Mutable, QuietCloseable, HttpRequestHeader {
     private final ObjectPool<DirectUtf8String> csPool;
     private final LowerCaseUtf8SequenceObjHashMap<DirectUtf8String> headers = new LowerCaseUtf8SequenceObjHashMap<>();
-    private final DirectUtf8Sink sink = new DirectUtf8Sink(0);
+    // Deliberately not a field initialiser: those run before the constructor body, so a throw from
+    // the header-buffer malloc would strand the sink with nothing able to close it. The constructor
+    // allocates it inside its own try instead.
+    private final DirectUtf8Sink sink;
     private final DirectUtf8String temp = new DirectUtf8String();
     private final Utf8SequenceObjHashMap<DirectUtf8String> urlParams = new Utf8SequenceObjHashMap<>();
     protected boolean incomplete;
@@ -73,10 +76,32 @@ public class HttpHeaderParser implements Mutable, QuietCloseable, HttpRequestHea
     private DirectUtf8String statusCode;
 
     public HttpHeaderParser(int bufferSize, ObjectPool<DirectUtf8String> csPool) {
-        this.headerPtr = this._wptr = Unsafe.malloc(bufferSize, MemoryTag.NATIVE_HTTP_CONN);
-        this.hi = headerPtr + bufferSize;
-        this.csPool = csPool;
-        clear();
+        // A local mirrors the sink field: the catch cannot read a blank final that the failing
+        // statement never assigned, so it frees this instead.
+        DirectUtf8Sink utf8Sink = null;
+        try {
+            this.sink = utf8Sink = new DirectUtf8Sink(0);
+            this.csPool = csPool;
+            this.headerPtr = this._wptr = Unsafe.malloc(bufferSize, MemoryTag.NATIVE_HTTP_CONN);
+            this.hi = headerPtr + bufferSize;
+            clear();
+        } catch (Throwable th) {
+            // Both native allocations sit inside this try, so the catch really can free every block
+            // the constructor managed to take. Unsafe.malloc throws whenever the JVM cannot hand out
+            // the block, and DirectUtf8Sink allocates even at capacity 0, so the caller can be left
+            // dropping a half-built parser that nothing will ever close. Free by hand rather than
+            // through close(): HttpClient.ResponseHeaders overrides close() to keep parser memory
+            // alive for the client to free later and to disconnect the outer client's socket, so
+            // routing this path through it would dispatch into a subclass mid-construction, tear
+            // down a live socket, and skip the frees entirely.
+            if (headerPtr != 0) {
+                headerPtr = _wptr = hi = Unsafe.free(headerPtr, hi - headerPtr, MemoryTag.NATIVE_HTTP_CONN);
+            }
+            if (utf8Sink != null) {
+                utf8Sink.close();
+            }
+            throw th;
+        }
     }
 
     @Override

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

@@ -83,16 +83,47 @@ public abstract class HttpClient implements QuietCloseable {
 
     public HttpClient(HttpClientConfiguration configuration, SocketFactory socketFactory) {
         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();
-        this.fixBrokenConnection = configuration.fixBrokenConnection();
-        this.bufLo = Unsafe.malloc(bufferSize, MemoryTag.NATIVE_DEFAULT);
-        this.responseParserBufLo = Unsafe.malloc(responseParserBufSize, MemoryTag.NATIVE_DEFAULT);
-        this.responseHeaders = new ResponseHeaders(responseParserBufLo, responseParserBufSize, defaultTimeout, 4096, csPool);
+        // Locals mirror the resources the constructor takes. A throw past the first of them - the
+        // ResponseHeaders parser allocates natively, and so do both mallocs - leaves a half-built
+        // client that no caller can reach, so close() will never run and the catch has to free
+        // what was taken. The catch cannot read a blank final the failing statement never
+        // assigned, hence the locals. Sizes are read up front for the same reason.
+        final int requestBufSize = configuration.getInitialRequestBufferSize();
+        final int responseBufSize = configuration.getResponseBufferSize();
+        Socket socket = null;
+        long requestBuf = 0;
+        long responseBuf = 0;
+        try {
+            this.socket = socket = socketFactory.newInstance(nf, LOG);
+            this.defaultTimeout = configuration.getTimeout();
+            this.connectTimeout = configuration.getConnectTimeout();
+            this.bufferSize = requestBufSize;
+            this.maxBufferSize = configuration.getMaximumRequestBufferSize();
+            this.responseParserBufSize = responseBufSize;
+            this.fixBrokenConnection = configuration.fixBrokenConnection();
+            this.bufLo = requestBuf = Unsafe.malloc(requestBufSize, MemoryTag.NATIVE_DEFAULT);
+            this.responseParserBufLo = responseBuf = Unsafe.malloc(responseBufSize, MemoryTag.NATIVE_DEFAULT);
+            this.responseHeaders = new ResponseHeaders(responseBuf, responseBufSize, defaultTimeout, 4096, csPool);
+        } catch (Throwable th) {
+            if (responseBuf != 0) {
+                this.responseParserBufLo = Unsafe.free(responseBuf, responseBufSize, MemoryTag.NATIVE_DEFAULT);
+            }
+            if (requestBuf != 0) {
+                this.bufLo = Unsafe.free(requestBuf, requestBufSize, MemoryTag.NATIVE_DEFAULT);
+            }
+            if (socket != null) {
+                // Attach a close failure to the primary one rather than let it replace it: a TLS
+                // socket can throw from close(), and the caller has to see why construction failed.
+                try {
+                    socket.close();
+                } catch (Throwable closeFailure) {
+                    if (closeFailure != th) {
+                        th.addSuppressed(closeFailure);
+                    }
+                }
+            }
+            throw th;
+        }
     }
 
     @Override
@@ -853,9 +884,28 @@ public class ResponseHeaders extends HttpHeaderParser {
 
         public ResponseHeaders(long respParserBufLo, int respParserBufSize, int defaultTimeout, int headerBufSize, ObjectPool<DirectUtf8String> pool) {
             super(headerBufSize, pool);
-            this.defaultTimeout = defaultTimeout;
-            this.response = new ResponseImpl(respParserBufLo, respParserBufLo + respParserBufSize, defaultTimeout);
-            this.chunkedResponse = new ChunkedResponseImpl(respParserBufLo, respParserBufLo + respParserBufSize, defaultTimeout);
+            try {
+                this.defaultTimeout = defaultTimeout;
+                this.response = new ResponseImpl(respParserBufLo, respParserBufLo + respParserBufSize, defaultTimeout);
+                this.chunkedResponse = new ChunkedResponseImpl(respParserBufLo, respParserBufLo + respParserBufSize, defaultTimeout);
+            } catch (Throwable th) {
+                // super() has already taken the parser's native header buffer and sink. The outer
+                // HttpClient never receives this object when a statement here throws, so its own
+                // catch cannot reach the parser and nothing would ever free it. Java does not run a
+                // superclass close() when a subclass constructor fails.
+                //
+                // The two response objects allocate nothing native, so only a Java-heap
+                // OutOfMemoryError reaches this catch, and there is no seam that can inject one
+                // without adding test-only production surface. That leaves the block untested by
+                // design; it is the same shape as the platform subclasses, which
+                // HttpClientConstructorTest does cover.
+                //
+                // super.close() rather than close(): close() is overridden to keep parser memory
+                // alive for the client to release later, and it would also tear down the outer
+                // client's live socket.
+                super.close();
+                throw th;
+            }
         }
 
         public void await() {

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

@@ -36,10 +36,19 @@ public class HttpClientLinux extends HttpClient {
 
     public HttpClientLinux(HttpClientConfiguration configuration, SocketFactory socketFactory) {
         super(configuration, socketFactory);
-        epoll = new Epoll(
-                configuration.getEpollFacade(),
-                configuration.getWaitQueueCapacity()
-        );
+        try {
+            epoll = new Epoll(
+                    configuration.getEpollFacade(),
+                    configuration.getWaitQueueCapacity()
+            );
+        } catch (Throwable th) {
+            // super() has already taken the socket, both buffers and the response parser. A throw
+            // here leaves a half-built client the caller never receives, so nothing would close it.
+            // super.close() rather than close(): epoll is still null and only the base class holds
+            // anything to release.
+            super.close();
+            throw th;
+        }
     }
 
     @Override

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

@@ -35,10 +35,19 @@ public class HttpClientOsx extends HttpClient {
 
     public HttpClientOsx(HttpClientConfiguration configuration, SocketFactory socketFactory) {
         super(configuration, socketFactory);
-        this.kqueue = new Kqueue(
-                configuration.getKQueueFacade(),
-                configuration.getWaitQueueCapacity()
-        );
+        try {
+            this.kqueue = new Kqueue(
+                    configuration.getKQueueFacade(),
+                    configuration.getWaitQueueCapacity()
+            );
+        } catch (Throwable th) {
+            // super() has already taken the socket, both buffers and the response parser. A throw
+            // here leaves a half-built client the caller never receives, so nothing would close it.
+            // super.close() rather than close(): kqueue is still null and only the base class holds
+            // anything to release.
+            super.close();
+            throw th;
+        }
     }
 
     @Override

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

@@ -37,8 +37,19 @@ public class HttpClientWindows extends HttpClient {
 
     public HttpClientWindows(HttpClientConfiguration configuration, SocketFactory socketFactory) {
         super(configuration, socketFactory);
-        this.fdSet = new FDSet(configuration.getWaitQueueCapacity());
-        this.sf = configuration.getSelectFacade();
+        try {
+            // Read the facade before taking the FD set, so nothing between the acquisition and the
+            // end of the constructor can throw and the catch has only the base class to release.
+            this.sf = configuration.getSelectFacade();
+            this.fdSet = new FDSet(configuration.getWaitQueueCapacity());
+        } catch (Throwable th) {
+            // super() has already taken the socket, both buffers and the response parser. A throw
+            // here leaves a half-built client the caller never receives, so nothing would close it.
+            // super.close() rather than close(): fdSet is still null and only the base class holds
+            // anything to release.
+            super.close();
+            throw th;
+        }
     }
 
     @Override

core/src/main/java/io/questdb/client/cutlass/line/http/AbstractLineHttpSender.java

@@ -192,15 +192,38 @@ protected AbstractLineHttpSender(
 
         this.isTls = tlsConfig != null;
 
-        if (client != null) {
-            this.client = client;
-        } else {
-            this.client = isTls ?
-                    HttpClientFactory.newTlsInstance(clientConfiguration, tlsConfig)
-                    : HttpClientFactory.newPlainTextInstance(clientConfiguration);
+        // The sender owns the client from the assignment on, whether the caller handed it in or
+        // this constructor built it - close() frees it either way. newRequest() writes the request
+        // preamble into the client's request buffer and throws when the path, the User-Agent
+        // header or an auth header does not fit, so a throw past the assignment would strand a
+        // socket and two native buffers that no caller ever receives a reference to.
+        // createLineSender() cannot clean that up on our behalf: it builds the sender in a return
+        // expression, past every catch it has.
+        try {
+            if (client != null) {
+                this.client = client;
+            } else {
+                this.client = isTls ?
+                        HttpClientFactory.newTlsInstance(clientConfiguration, tlsConfig)
+                        : HttpClientFactory.newPlainTextInstance(clientConfiguration);
+            }
+            this.questDBVersion = new BuildInformationHolder().getSwVersion();
+            this.request = newRequest();
+        } catch (Throwable th) {
+            if (this.client != null) {
+                // Attach a close failure to the primary one rather than let it replace it: a TLS
+                // socket can throw from close(), and the caller has to see why construction failed.
+                try {
+                    this.client.close();
+                } catch (Throwable closeFailure) {
+                    if (closeFailure != th) {
+                        th.addSuppressed(closeFailure);
+                    }
+                }
+                this.client = null;
+            }
+            throw th;
         }
-        this.questDBVersion = new BuildInformationHolder().getSwVersion();
-        this.request = newRequest();
         this.maxNameLength = maxNameLength;
         this.rnd = rnd;
     }

core/src/main/java/io/questdb/client/impl/QueryClientPool.java

@@ -97,6 +97,10 @@ public final class QueryClientPool implements AutoCloseable {
     // DEFAULT_CLOSE_QUERY_TIMEOUT_MILLIS. Volatile because QuestDBImpl sets it
     // once at build time on a different thread than the borrowers that read it.
     private volatile long closeQueryTimeoutMillis = DEFAULT_CLOSE_QUERY_TIMEOUT_MILLIS;
+    // Threads currently inside close()'s bounded creation-wait loop. Exists so
+    // hasCreationWaiterForTesting() can report that region as a stable state
+    // rather than probing the condition queue; see that method. Guarded by lock.
+    private int closeCreationWaiters;
     private int inFlightCreations;
 
     public QueryClientPool(
@@ -339,13 +343,18 @@ public void close() {
             final long creationWaitDeadlineNanos = System.nanoTime() + creationWaitNanos;
             long creationRemainingNanos = creationWaitNanos;
             boolean creationWaitInterrupted = false;
-            while (inFlightCreations > 0 && creationRemainingNanos > 0) {
-                try {
-                    creationFinished.awaitNanos(creationRemainingNanos);
-                } catch (InterruptedException e) {
-                    creationWaitInterrupted = true;
+            closeCreationWaiters++;
+            try {
+                while (inFlightCreations > 0 && creationRemainingNanos > 0) {
+                    try {
+                        creationFinished.awaitNanos(creationRemainingNanos);
+                    } catch (InterruptedException e) {
+                        creationWaitInterrupted = true;
+                    }
+                    creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime();
                 }
-                creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime();
+            } finally {
+                closeCreationWaiters--;
             }
             if (creationWaitInterrupted) {
                 Thread.currentThread().interrupt();
@@ -528,11 +537,17 @@ void release(QueryWorker w, long gen) {
         }
     }
 
+    // True while a close() is inside its bounded creation wait. Deliberately not
+    // lock.hasWaiters(creationFinished): an interrupt moves the waiter out of the
+    // condition queue and into the lock's sync queue to reacquire before await
+    // rethrows, so under an interrupt storm hasWaiters() reads false for a
+    // measurable fraction of the wait even though close() never left it. Tests
+    // that assert on the wait region need a state that does not flicker.
     @TestOnly
     public boolean hasCreationWaiterForTesting() {
         lock.lock();
         try {
-            return lock.hasWaiters(creationFinished);
+            return closeCreationWaiters > 0;
         } finally {
             lock.unlock();
         }

core/src/main/java/io/questdb/client/impl/SenderPool.java

@@ -238,6 +238,10 @@ public final class SenderPool implements AutoCloseable {
     // (cancelling recovery) WITHOUT making a later close() short-circuit the
     // teardown. Guarded by lock.
     private boolean closeStarted;
+    // Threads currently inside close()'s bounded creation-wait loop. Exists so
+    // hasCreationWaiterForTesting() can report that region as a stable state
+    // rather than probing the condition queue; see that method. Guarded by lock.
+    private int closeCreationWaiters;
     private int inFlightCreations;
     // Lease teardowns currently running on borrower threads (retireLease's
     // delegate-close section, outside the lock). close() counts these as
@@ -1339,11 +1343,17 @@ public Thread getStartupRecoveryThreadForTesting() {
         return startupRecoveryThread;
     }
 
+    // True while a close() is inside its bounded creation wait. Deliberately not
+    // lock.hasWaiters(creationFinished): an interrupt moves the waiter out of the
+    // condition queue and into the lock's sync queue to reacquire before await
+    // rethrows, so under an interrupt storm hasWaiters() reads false for a
+    // measurable fraction of the wait even though close() never left it. Tests
+    // that assert on the wait region need a state that does not flicker.
     @TestOnly
     public boolean hasCreationWaiterForTesting() {
         lock.lock();
         try {
-            return lock.hasWaiters(creationFinished);
+            return closeCreationWaiters > 0;
         } finally {
             lock.unlock();
         }
@@ -1483,13 +1493,18 @@ public void close() {
             final long creationWaitDeadlineNanos = System.nanoTime() + creationWaitNanos;
             long creationRemainingNanos = creationWaitNanos;
             boolean creationWaitInterrupted = false;
-            while (inFlightCreations > 0 && creationRemainingNanos > 0) {
-                try {
-                    creationFinished.awaitNanos(creationRemainingNanos);
-                } catch (InterruptedException e) {
-                    creationWaitInterrupted = true;
+            closeCreationWaiters++;
+            try {
+                while (inFlightCreations > 0 && creationRemainingNanos > 0) {
+                    try {
+                        creationFinished.awaitNanos(creationRemainingNanos);
+                    } catch (InterruptedException e) {
+                        creationWaitInterrupted = true;
+                    }
+                    creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime();
                 }
-                creationRemainingNanos = creationWaitDeadlineNanos - System.nanoTime();
+            } finally {
+                closeCreationWaiters--;
             }
             if (creationWaitInterrupted) {
                 Thread.currentThread().interrupt();