From d5837915bc23288f8a41fc7297070b0b0b0fa47f Mon Sep 17 00:00:00 2001 From: Andrei Pechkurov Date: Wed, 29 Jul 2026 20:32:47 +0300 Subject: [PATCH 1/3] Roll back HTTP client constructors on failure The ILP HTTP senders build their client through this fork's HttpClientFactory, whose constructors took a socket and native buffers with no rollback. A throw past the first acquisition left a half-built client the caller never receives, so close() never ran and everything taken so far leaked, the socket file descriptor included. HttpClient now takes the socket, both buffers and the response parser inside one try and frees what it took in reverse order, attaching a close failure to the primary exception rather than replacing it. ResponseHeaders does the same for the parser its super() call completed. HttpHeaderParser moves the direct UTF-8 sink out of its field initialiser, which runs before the constructor body, so the constructor's own catch can reach it. The three platform subclasses close the completed base client when their poller fails to open. HttpClientWindows also reads the select facade before it takes the FD set. Nothing between that acquisition and the end of the constructor can then throw, which removes a partial-FDSet cleanup branch that only a Windows host could ever have reached. HttpClientConstructorTest covers each rollback point with a close-counting socket, and HttpHeaderParserTest covers the sink. Both inject the fault deterministically and without a host dependency: a negative buffer size makes Unsafe.allocateMemory throw, and a throwing configuration getter fails each platform subclass before it touches platform natives. Co-Authored-By: Claude Opus 5 (1M context) --- .../client/cutlass/http/HttpHeaderParser.java | 35 +++- .../cutlass/http/client/HttpClient.java | 76 +++++-- .../cutlass/http/client/HttpClientLinux.java | 17 +- .../cutlass/http/client/HttpClientOsx.java | 17 +- .../http/client/HttpClientWindows.java | 15 +- .../cutlass/http/HttpHeaderParserTest.java | 24 +++ .../client/HttpClientConstructorTest.java | 192 ++++++++++++++++++ 7 files changed, 348 insertions(+), 28 deletions(-) create mode 100644 core/src/test/java/io/questdb/client/test/cutlass/http/client/HttpClientConstructorTest.java diff --git a/core/src/main/java/io/questdb/client/cutlass/http/HttpHeaderParser.java b/core/src/main/java/io/questdb/client/cutlass/http/HttpHeaderParser.java index 6433c782..ee132778 100644 --- a/core/src/main/java/io/questdb/client/cutlass/http/HttpHeaderParser.java +++ b/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 csPool; private final LowerCaseUtf8SequenceObjHashMap 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 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 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 diff --git a/core/src/main/java/io/questdb/client/cutlass/http/client/HttpClient.java b/core/src/main/java/io/questdb/client/cutlass/http/client/HttpClient.java index 0175ad6c..a915340a 100644 --- a/core/src/main/java/io/questdb/client/cutlass/http/client/HttpClient.java +++ b/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 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() { diff --git a/core/src/main/java/io/questdb/client/cutlass/http/client/HttpClientLinux.java b/core/src/main/java/io/questdb/client/cutlass/http/client/HttpClientLinux.java index 472b5257..0f132d92 100644 --- a/core/src/main/java/io/questdb/client/cutlass/http/client/HttpClientLinux.java +++ b/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 diff --git a/core/src/main/java/io/questdb/client/cutlass/http/client/HttpClientOsx.java b/core/src/main/java/io/questdb/client/cutlass/http/client/HttpClientOsx.java index aae49dc3..55491f01 100644 --- a/core/src/main/java/io/questdb/client/cutlass/http/client/HttpClientOsx.java +++ b/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 diff --git a/core/src/main/java/io/questdb/client/cutlass/http/client/HttpClientWindows.java b/core/src/main/java/io/questdb/client/cutlass/http/client/HttpClientWindows.java index 62ee43fe..49a611db 100644 --- a/core/src/main/java/io/questdb/client/cutlass/http/client/HttpClientWindows.java +++ b/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 diff --git a/core/src/test/java/io/questdb/client/test/cutlass/http/HttpHeaderParserTest.java b/core/src/test/java/io/questdb/client/test/cutlass/http/HttpHeaderParserTest.java index 6974d0b5..76268a31 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/http/HttpHeaderParserTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/http/HttpHeaderParserTest.java @@ -27,6 +27,7 @@ import io.questdb.client.cutlass.http.HttpException; import io.questdb.client.cutlass.http.HttpHeaderParser; import io.questdb.client.std.MemoryTag; +import io.questdb.client.std.Misc; import io.questdb.client.std.ObjectPool; import io.questdb.client.std.Rnd; import io.questdb.client.std.Unsafe; @@ -52,6 +53,29 @@ public class HttpHeaderParserTest { "Cookie: textwrapon=false; textautoformat=false; wysiwyg=textarea\r\n" + "\r\n"; + @Test + public void testConstructorFailureFreesNativeAllocations() throws Exception { + // The constructor takes the sink first and then mallocs the header buffer. Nothing ever + // closes a parser whose constructor threw, so the catch has to release the sink. A negative + // buffer size is how the client can fail that malloc deterministically: unlike the server it + // has no RSS-limit seam, and sun.misc.Unsafe.allocateMemory rejects a negative size with + // IllegalArgumentException on every supported JDK. + assertMemoryLeak(() -> { + // Holds the parser on the path where the constructor unexpectedly succeeds. Dropping it + // there would leak a built parser and make the enclosing leak check fail on top of the + // Assert.fail below, burying the failure that matters. + HttpHeaderParser parser = null; + try { + parser = new HttpHeaderParser(-1, pool); + Assert.fail("expected IllegalArgumentException"); + } catch (IllegalArgumentException ignore) { + // the header-buffer malloc rejected the negative size + } finally { + Misc.free(parser); + } + }); + } + @Test public void testContentLengthLarge() throws Exception { assertMemoryLeak(() -> { diff --git a/core/src/test/java/io/questdb/client/test/cutlass/http/client/HttpClientConstructorTest.java b/core/src/test/java/io/questdb/client/test/cutlass/http/client/HttpClientConstructorTest.java new file mode 100644 index 00000000..2e7a2402 --- /dev/null +++ b/core/src/test/java/io/questdb/client/test/cutlass/http/client/HttpClientConstructorTest.java @@ -0,0 +1,192 @@ +/*+***************************************************************************** + * ___ _ ____ ____ + * / _ \ _ _ ___ ___| |_| _ \| __ ) + * | | | | | | |/ _ \/ __| __| | | | _ \ + * | |_| | |_| | __/\__ \ |_| |_| | |_) | + * \__\_\\__,_|\___||___/\__|____/|____/ + * + * 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.http.client; + +import io.questdb.client.DefaultHttpClientConfiguration; +import io.questdb.client.HttpClientConfiguration; +import io.questdb.client.cutlass.http.client.HttpClient; +import io.questdb.client.cutlass.http.client.HttpClientFactory; +import io.questdb.client.cutlass.http.client.HttpClientLinux; +import io.questdb.client.cutlass.http.client.HttpClientOsx; +import io.questdb.client.cutlass.http.client.HttpClientWindows; +import io.questdb.client.network.EpollFacade; +import io.questdb.client.network.KqueueFacade; +import io.questdb.client.network.NetworkFacade; +import io.questdb.client.network.PlainSocket; +import io.questdb.client.network.SelectFacade; +import io.questdb.client.network.Socket; +import io.questdb.client.network.SocketFactory; +import io.questdb.client.test.tools.TestUtils; +import org.junit.Assert; +import org.junit.Test; +import org.slf4j.Logger; + +/** + * Covers the rollback the HTTP client constructors run when a later acquisition throws. A + * half-built client never reaches the caller, so nothing will ever close it and the constructor + * itself has to release what it already took. The enclosing {@code assertMemoryLeak} observes the + * native blocks, and the close counter on the injected socket observes the socket. + */ +public class HttpClientConstructorTest { + + @Test + public void testBaseConstructorFailureAfterRequestBufferReleasesEverything() throws Exception { + // A negative response-buffer size makes the second malloc throw with the socket and the + // request buffer already taken. sun.misc.Unsafe.allocateMemory rejects a negative size with + // IllegalArgumentException on every supported JDK, which makes this the one fault the + // client can inject deterministically: it has no RSS-limit seam the way the server does. + final HttpClientConfiguration configuration = new DefaultHttpClientConfiguration() { + @Override + public int getInitialRequestBufferSize() { + return 1024; + } + + @Override + public int getResponseBufferSize() { + return -1; + } + }; + + TestUtils.assertMemoryLeak(() -> { + final CountingSocketFactory socketFactory = new CountingSocketFactory(); + try { + buildAndClose(() -> HttpClientFactory.newInstance(configuration, socketFactory)); + Assert.fail("expected IllegalArgumentException"); + } catch (IllegalArgumentException ignore) { + // the response-buffer malloc rejected the negative size + } + Assert.assertEquals("the constructor must close the socket it took", 1, socketFactory.closeCount); + }); + } + + @Test + public void testBaseConstructorFailureAfterSocketClosesSocket() throws Exception { + // getTimeout() is the first configuration read after the socket, so this fails with the + // socket taken and no native block allocated yet. + final HttpClientConfiguration configuration = new DefaultHttpClientConfiguration() { + @Override + public int getTimeout() { + throw new InjectedFailure(); + } + }; + + assertInjectedFailureRollback(socketFactory -> HttpClientFactory.newInstance(configuration, socketFactory)); + } + + @Test + public void testLinuxConstructorFailureClosesBaseClient() throws Exception { + // super() has completed by the time the subclass runs, so the socket, both buffers and the + // response parser are all live and the subclass catch has to hand them all back. The facade + // getter throws before new Epoll(...) is invoked, so the test never touches epoll and is + // host-independent. + final HttpClientConfiguration configuration = new DefaultHttpClientConfiguration() { + @Override + public EpollFacade getEpollFacade() { + throw new InjectedFailure(); + } + }; + + assertInjectedFailureRollback(socketFactory -> new HttpClientLinux(configuration, socketFactory)); + } + + @Test + public void testOsxConstructorFailureClosesBaseClient() throws Exception { + final HttpClientConfiguration configuration = new DefaultHttpClientConfiguration() { + @Override + public KqueueFacade getKQueueFacade() { + throw new InjectedFailure(); + } + }; + + assertInjectedFailureRollback(socketFactory -> new HttpClientOsx(configuration, socketFactory)); + } + + @Test + public void testWindowsConstructorFailureClosesBaseClient() throws Exception { + // The Windows constructor reads the select facade before it takes the FD set, so this + // failure lands with only the base class holding resources - and, unlike a failing + // new FDSet(...), it never initialises the Windows-only SelectAccessor natives. + final HttpClientConfiguration configuration = new DefaultHttpClientConfiguration() { + @Override + public SelectFacade getSelectFacade() { + throw new InjectedFailure(); + } + }; + + assertInjectedFailureRollback(socketFactory -> new HttpClientWindows(configuration, socketFactory)); + } + + private static void assertInjectedFailureRollback(ClientFactory clientFactory) throws Exception { + TestUtils.assertMemoryLeak(() -> { + final CountingSocketFactory socketFactory = new CountingSocketFactory(); + try { + buildAndClose(() -> clientFactory.newInstance(socketFactory)); + Assert.fail("expected InjectedFailure"); + } catch (InjectedFailure ignore) { + // the injected configuration read threw + } + Assert.assertEquals("the constructor must close the socket it took", 1, socketFactory.closeCount); + }); + } + + private static void buildAndClose(ClientSupplier supplier) { + // Closing here keeps the leak check honest on the path where the constructor unexpectedly + // succeeds: dropping a built client would leak on top of the failure the caller asserts and + // bury it. + HttpClient client = supplier.get(); + client.close(); + } + + @FunctionalInterface + private interface ClientFactory { + HttpClient newInstance(SocketFactory socketFactory); + } + + @FunctionalInterface + private interface ClientSupplier { + HttpClient get(); + } + + private static class CountingSocketFactory implements SocketFactory { + int closeCount; + + @Override + public Socket newInstance(NetworkFacade nf, Logger log) { + return new PlainSocket(nf, log) { + @Override + public synchronized void close() { + closeCount++; + super.close(); + } + }; + } + } + + private static class InjectedFailure extends RuntimeException { + InjectedFailure() { + super("injected constructor failure", null, false, false); + } + } +} From 7d420f7d17932f10e37dca53e4649270b829f0ea Mon Sep 17 00:00:00 2001 From: Andrei Pechkurov Date: Thu, 30 Jul 2026 11:29:09 +0300 Subject: [PATCH 2/3] Close the HTTP client when sender startup fails AbstractLineHttpSender's constructor assigned this.client - either the one createLineSender() handed in or one it built through HttpClientFactory - and then called newRequest(), with neither statement inside a try. newRequest() writes the POST line, the path, the User-Agent header and any auth header into the client's request buffer, and growBuffer() throws once they exceed the maximum. A throw there left a fully built client - a socket plus two native buffers plus the response parser - that no caller ever receives a reference to, so nothing closed it. A Java-heap OutOfMemoryError from the version lookup or the User-Agent concatenation lands in the same window. createLineSender() cannot clean this up. Its three Misc.free(cli) calls all sit before the switch, and the switch builds the sender inside a return expression that no try covers. Both entry shapes leak: an explicit protocol version makes the constructor build the client itself, and an auto-detected one hands in the client the version probe used. The constructor now takes the client, the version and the request inside one try and closes the client on the way out, attaching a close failure to the primary exception rather than replacing it. It frees a handed-in client too: close() already frees the client regardless of where it came from, so the sender owns it from the assignment on either way. LineHttpSenderConstructorTest covers both shapes. A configuration whose maximum request buffer is smaller than the preamble makes newRequest() throw deterministically, with no server and no new production seam. The self-built case asserts against the enclosing leak check; the handed-in case adds a close-counting socket, since the leak check cannot see a file descriptor. Both fail without the fix: the first on 65552 bytes leaked under NATIVE_DEFAULT, the second on the missing socket close. Co-Authored-By: Claude Opus 5 (1M context) --- .../line/http/AbstractLineHttpSender.java | 39 +++- .../line/LineHttpSenderConstructorTest.java | 168 ++++++++++++++++++ 2 files changed, 199 insertions(+), 8 deletions(-) create mode 100644 core/src/test/java/io/questdb/client/test/cutlass/line/LineHttpSenderConstructorTest.java diff --git a/core/src/main/java/io/questdb/client/cutlass/line/http/AbstractLineHttpSender.java b/core/src/main/java/io/questdb/client/cutlass/line/http/AbstractLineHttpSender.java index 398aa70a..268d271b 100644 --- a/core/src/main/java/io/questdb/client/cutlass/line/http/AbstractLineHttpSender.java +++ b/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; } diff --git a/core/src/test/java/io/questdb/client/test/cutlass/line/LineHttpSenderConstructorTest.java b/core/src/test/java/io/questdb/client/test/cutlass/line/LineHttpSenderConstructorTest.java new file mode 100644 index 00000000..c25724cc --- /dev/null +++ b/core/src/test/java/io/questdb/client/test/cutlass/line/LineHttpSenderConstructorTest.java @@ -0,0 +1,168 @@ +/*+***************************************************************************** + * ___ _ ____ ____ + * / _ \ _ _ ___ ___| |_| _ \| __ ) + * | | | | | | |/ _ \/ __| __| | | | _ \ + * | |_| | |_| | __/\__ \ |_| |_| | |_) | + * \__\_\\__,_|\___||___/\__|____/|____/ + * + * 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.line; + +import io.questdb.client.DefaultHttpClientConfiguration; +import io.questdb.client.HttpClientConfiguration; +import io.questdb.client.Sender; +import io.questdb.client.cutlass.http.client.HttpClient; +import io.questdb.client.cutlass.http.client.HttpClientException; +import io.questdb.client.cutlass.http.client.HttpClientFactory; +import io.questdb.client.cutlass.line.http.AbstractLineHttpSender; +import io.questdb.client.cutlass.line.http.LineHttpSenderV2; +import io.questdb.client.network.NetworkFacade; +import io.questdb.client.network.PlainSocket; +import io.questdb.client.network.Socket; +import io.questdb.client.network.SocketFactory; +import io.questdb.client.std.IntList; +import io.questdb.client.std.Misc; +import io.questdb.client.std.ObjList; +import io.questdb.client.std.Rnd; +import io.questdb.client.test.tools.TestUtils; +import org.junit.Assert; +import org.junit.Test; +import org.slf4j.Logger; + +/** + * Covers the rollback the sender constructor runs when {@code newRequest()} throws. The sender owns + * its client from the assignment on - {@code close()} frees it whether the caller handed it in or + * the constructor built it - but a failed constructor hands no reference back, so nothing else can + * close it. {@link AbstractLineHttpSender#createLineSender} cannot cover this either: it builds the + * sender in a return expression, past every catch it has. + */ +public class LineHttpSenderConstructorTest { + // Smaller than "POST /write HTTP/1.1\r\n", so newRequest() cannot fit the preamble and throws + // with the client fully built. A third-party HttpClientConfiguration can reach this in + // production; LineSenderBuilder floors the maximum at 64 KiB, which puts a heap + // OutOfMemoryError between the assignment and newRequest() in the same window. + private static final int TINY_BUFFER_SIZE = 16; + private static final HttpClientConfiguration TINY_BUFFER_CONFIGURATION = new DefaultHttpClientConfiguration() { + @Override + public int getInitialRequestBufferSize() { + return TINY_BUFFER_SIZE; + } + + @Override + public int getMaximumRequestBufferSize() { + return TINY_BUFFER_SIZE; + } + }; + + @Test + public void testHandedInClientIsReleasedWhenRequestPreambleDoesNotFit() throws Exception { + // The auto-detecting createLineSender() builds the client itself and hands it to the + // constructor, so the constructor has to free a client it did not create. The counting + // socket observes the file descriptor the leak check cannot see. + TestUtils.assertMemoryLeak(() -> { + final CountingSocketFactory socketFactory = new CountingSocketFactory(); + final HttpClient client = HttpClientFactory.newInstance(TINY_BUFFER_CONFIGURATION, socketFactory); + try { + buildAndClose(newSender(client)); + Assert.fail("expected HttpClientException"); + } catch (HttpClientException ignore) { + // newRequest() could not fit the preamble + } + // Two closes: newRequest() disconnects first because the sender's host differs from the + // client's, then the rollback closes the client itself. Without the rollback the count + // is 1 - and the client's native buffers stay live for the enclosing leak check. + Assert.assertEquals("the constructor must close the client it was handed", 2, socketFactory.closeCount); + }); + } + + @Test + public void testSelfBuiltClientIsReleasedWhenRequestPreambleDoesNotFit() throws Exception { + // An explicit protocol version skips detection, so createLineSender() passes a null client + // and the constructor builds its own. HttpClientFactory pins PlainSocketFactory on that + // path, so the leak check on the client's native buffers is the link here. + TestUtils.assertMemoryLeak(() -> { + try { + buildAndClose(AbstractLineHttpSender.createLineSender( + new ObjList<>("localhost"), + IntList.createWithValues(9000), + "/write", + TINY_BUFFER_CONFIGURATION, + null, + 1000, + null, + null, + null, + 127, + 0, + 0, + 0, + Long.MAX_VALUE, + Sender.PROTOCOL_VERSION_V2 + )); + Assert.fail("expected HttpClientException"); + } catch (HttpClientException ignore) { + // newRequest() could not fit the preamble + } + }); + } + + private static void buildAndClose(Sender sender) { + // Closing here keeps the leak check honest on the path where the constructor unexpectedly + // succeeds: dropping a built sender would leak on top of the failure the caller asserts and + // bury it. + Misc.free(sender); + } + + private static LineHttpSenderV2 newSender(HttpClient client) { + return new LineHttpSenderV2( + new ObjList<>("localhost"), + IntList.createWithValues(9000), + "/write", + TINY_BUFFER_CONFIGURATION, + null, + client, + 1000, + null, + null, + null, + 127, + 0, + 0, + 0, + Long.MAX_VALUE, + 0, + new Rnd() + ); + } + + private static class CountingSocketFactory implements SocketFactory { + int closeCount; + + @Override + public Socket newInstance(NetworkFacade nf, Logger log) { + return new PlainSocket(nf, log) { + @Override + public synchronized void close() { + closeCount++; + super.close(); + } + }; + } + } +} From 0ddfc915c5e40a0d88813e929fa8efd98da50c5b Mon Sep 17 00:00:00 2001 From: Andrei Pechkurov Date: Thu, 30 Jul 2026 13:19:23 +0300 Subject: [PATCH 3/3] test: observe the close creation-wait as state, not a condition-queue probe facadeCloseIsBoundedUnderRepeatedInterruptsDuringSenderCreation failed on windows-msvc-2022-x64 with "sender close left its creation wait before the interrupt storm landed twice; interrupts landed: 6" -- a false negative, not a product defect. Both pools' hasCreationWaiterForTesting() reported the wait region as lock.hasWaiters(creationFinished). An interrupt does not leave the waiter on the condition queue: AQS transfers the node to the lock's sync queue to reacquire before awaitNanos rethrows, so under the test's interrupt storm the probe reads false for a measurable slice of a wait close() never left. A harness replicating close()'s loop measures 5-8% of polls reading false with the 1s deadline honoured exactly. Count the threads inside the wait loop instead: the state no longer flickers, and the rising edge no longer depends on the closer having actually parked. awaitRepeatedInterrupts() also re-reads the interrupt count before failing, since the count it samples precedes the predicate read that fails on it. creationWaitStateDoesNotFlickerUnderInterruptStorm pins the contract: red on the old predicate (flickered after 46 polls, 2 interrupts landed), green now. Full suite green locally: 2766 tests, 0 failures. Co-Authored-By: Claude Opus 5 (1M context) --- .../questdb/client/impl/QueryClientPool.java | 29 ++++-- .../io/questdb/client/impl/SenderPool.java | 29 ++++-- .../impl/QuestDBImplCloseLifecycleTest.java | 90 ++++++++++++++++++- 3 files changed, 133 insertions(+), 15 deletions(-) diff --git a/core/src/main/java/io/questdb/client/impl/QueryClientPool.java b/core/src/main/java/io/questdb/client/impl/QueryClientPool.java index 75323a12..cbfe2a0c 100644 --- a/core/src/main/java/io/questdb/client/impl/QueryClientPool.java +++ b/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(); } diff --git a/core/src/main/java/io/questdb/client/impl/SenderPool.java b/core/src/main/java/io/questdb/client/impl/SenderPool.java index 2d01059e..9e08b7ac 100644 --- a/core/src/main/java/io/questdb/client/impl/SenderPool.java +++ b/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(); diff --git a/core/src/test/java/io/questdb/client/test/impl/QuestDBImplCloseLifecycleTest.java b/core/src/test/java/io/questdb/client/test/impl/QuestDBImplCloseLifecycleTest.java index 6344336b..21e50ff7 100644 --- a/core/src/test/java/io/questdb/client/test/impl/QuestDBImplCloseLifecycleTest.java +++ b/core/src/test/java/io/questdb/client/test/impl/QuestDBImplCloseLifecycleTest.java @@ -351,6 +351,83 @@ public void facadeCloseIsBoundedUnderRepeatedInterruptsDuringSenderCreation() th }); } + /** + * Pins the contract {@link #awaitRepeatedInterrupts} depends on: while a close sits in its + * bounded creation wait, {@code hasCreationWaiterForTesting()} must stay true through an + * interrupt storm. Implementing it as {@code lock.hasWaiters(creationFinished)} does not: + * an interrupt moves the waiter off the condition queue and onto the lock's sync queue to + * reacquire before await rethrows, so the probe reads false for a few percent of the samples + * while the product is still honouring the very deadline the storm is meant to attack. That + * flicker fails the interrupt tests above on slower agents (seen on windows-msvc-2022-x64), + * blaming the product for a defect in the observation. + */ + @Test(timeout = 30_000) + public void creationWaitStateDoesNotFlickerUnderInterruptStorm() throws Exception { + TestUtils.assertMemoryLeak(() -> { + CountDownLatch inCreation = new CountDownLatch(1); + CountDownLatch releaseCreation = new CountDownLatch(1); + AtomicInteger teardownCount = new AtomicInteger(); + IntFunction senderFactory = slotIndex -> { + inCreation.countDown(); + awaitOrFail(releaseCreation, "test never released sender creation"); + return fakeSender(teardownCount, null, null); + }; + String senderConfig = "ws::addr=localhost:1;sf_dir=" + + System.getProperty("java.io.tmpdir") + "/qdb-stable-wait-pool-" + System.nanoTime() + ";"; + // The creation-wait budget is the full MAX_CLOSE_CREATION_WAIT_MILLIS cap so the poll + // window below cannot race a legitimate expiry even on a saturated agent: any false + // reading inside it is the flicker, not the deadline. + QuestDBImpl db = newQuestDB(senderConfig, 0, 0, 5000, senderFactory, client -> { + }); + SenderPool pool = db.getSenderPoolForTesting(); + AtomicBoolean keepInterrupting = new AtomicBoolean(true); + AtomicInteger interruptCount = new AtomicInteger(); + Thread borrower = new Thread(() -> { + try { + db.borrowSender(); + } catch (Throwable ignored) { + } + }, "stable-wait-sender-borrower"); + Thread closer = new Thread(db::close, "stable-wait-sender-closer"); + Thread interrupter = new Thread(() -> { + while (keepInterrupting.get()) { + interruptCount.incrementAndGet(); + closer.interrupt(); + Thread.yield(); + } + }, "stable-wait-close-interrupter"); + try { + borrower.start(); + Assert.assertTrue("sender borrow never reached construction", + inCreation.await(10, TimeUnit.SECONDS)); + + closer.start(); + awaitCreationWaiter(pool, + "facade close did not wait while sender construction was internally owned"); + interrupter.start(); + long until = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(300); + int polls = 0; + while (System.nanoTime() < until) { + if (!pool.hasCreationWaiterForTesting()) { + Assert.fail("creation-wait state flickered under the interrupt storm after " + + polls + " polls; interrupts landed: " + interruptCount.get()); + } + polls++; + Thread.yield(); + } + Assert.assertTrue("interrupt storm never landed", interruptCount.get() > 1); + Assert.assertTrue("close must still be inside its creation wait", closer.isAlive()); + } finally { + keepInterrupting.set(false); + releaseCreation.countDown(); + interrupter.join(TimeUnit.SECONDS.toMillis(10)); + db.close(); + borrower.join(TimeUnit.SECONDS.toMillis(10)); + closer.join(TimeUnit.SECONDS.toMillis(10)); + } + }); + } + @Test(timeout = 30_000) public void facadeCloseWaitsForQueryCreationAndTeardown() throws Exception { TestUtils.assertMemoryLeak(() -> { @@ -536,6 +613,11 @@ private static void awaitCreationWaiter(SenderPool pool, String message) { * nothing: with a post-join count assert alone, the run races the close budget against thread * scheduling and can fail with the product invariant intact. Failing here instead separates * "interrupter starved before the budget expired" from a genuine deadline bug. + *

+ * {@code closerStillWaiting} must report the wait region as a stable state, not probe the + * condition queue: every interrupt moves the closer out of that queue to reacquire the lock, + * so a queue probe reads "not waiting" for a few percent of the poll samples with the product + * invariant intact. Both pools' {@code hasCreationWaiterForTesting()} count the region instead. */ private static void awaitRepeatedInterrupts( AtomicInteger interruptCount, @@ -550,7 +632,13 @@ private static void awaitRepeatedInterrupts( return; } if (!closerStillWaiting.getAsBoolean()) { - Assert.fail(message + "; interrupts landed: " + interruptCount.get()); + // The count above was sampled before the predicate, so the second interrupt may + // have landed in between; re-read before failing rather than on a stale sample. + int landed = interruptCount.get(); + if (landed > 1) { + return; + } + Assert.fail(message + "; interrupts landed: " + landed); } Thread.yield(); }