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/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/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/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); + } + } +} 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(); + } + }; + } + } +} 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(); }