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 6433c7828..30effdc97 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 @@ -26,6 +26,7 @@ import io.questdb.client.std.LowerCaseUtf8SequenceObjHashMap; import io.questdb.client.std.MemoryTag; +import io.questdb.client.std.Misc; import io.questdb.client.std.Mutable; import io.questdb.client.std.Numbers; import io.questdb.client.std.NumericException; @@ -46,7 +47,8 @@ public class HttpHeaderParser implements Mutable, QuietCloseable, HttpRequestHeader { private final ObjectPool csPool; private final LowerCaseUtf8SequenceObjHashMap headers = new LowerCaseUtf8SequenceObjHashMap<>(); - private final DirectUtf8Sink sink = new DirectUtf8Sink(0); + // Allocate inside the constructor's try so later initialization failures release the sink. + private final DirectUtf8Sink sink; private final DirectUtf8String temp = new DirectUtf8String(); private final Utf8SequenceObjHashMap urlParams = new Utf8SequenceObjHashMap<>(); protected boolean incomplete; @@ -73,10 +75,17 @@ 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(); + try { + this.sink = 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) { + // Avoid invoking an overridden close() during construction. + freeNative(); + throw th; + } } @Override @@ -108,10 +117,7 @@ public void clear() { @Override public void close() { clear(); - if (headerPtr != 0) { - headerPtr = _wptr = hi = Unsafe.free(headerPtr, hi - headerPtr, MemoryTag.NATIVE_HTTP_CONN); - } - sink.close(); + freeNative(); csPool.clear(); } @@ -204,6 +210,13 @@ public long parse(long ptr, long hi, boolean _method, boolean _protocol) { return p; } + private void freeNative() { + if (headerPtr != 0) { + headerPtr = _wptr = hi = Unsafe.free(headerPtr, hi - headerPtr, MemoryTag.NATIVE_HTTP_CONN); + } + Misc.free(sink); + } + private void parseContentLength() { contentLength = -1; DirectUtf8Sequence seq = getHeader(HEADER_CONTENT_LENGTH); 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 941dc58ff..9d8fd7c86 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 @@ -111,7 +111,7 @@ public HttpClient(HttpClientConfiguration configuration, SocketFactory socketFac if (stagedBufLo != 0) { Unsafe.free(stagedBufLo, bufferSize, MemoryTag.NATIVE_DEFAULT); } - Misc.free(stagedSocket); + Misc.freeSuppressing(stagedSocket, t); throw t; } this.socket = stagedSocket; @@ -911,12 +911,7 @@ public class ResponseHeaders extends HttpHeaderParser { public ResponseHeaders(long respParserBufLo, int respParserBufSize, int defaultTimeout, int headerBufSize, ObjectPool pool) { super(headerBufSize, pool); - // super() mallocs the header parse buffer as its FIRST statement, so from here on this object owns - // native memory while still being unreachable by anyone who could free it. A heap OOM in either - // allocation below would strand those bytes past the enclosing constructor's catch (Throwable), - // which frees only what IT staged - it never holds a reference to a ResponseHeaders that failed - // to finish constructing. Same rule as out there: whoever took it frees it when construction - // cannot complete. + // Release native memory allocated by the superclass if subclass initialization fails. try { this.defaultTimeout = defaultTimeout; this.response = new ResponseImpl(respParserBufLo, respParserBufLo + respParserBufSize, defaultTimeout); 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 49204fa8e..eadc0495f 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 @@ -76,6 +76,7 @@ public abstract class AbstractLineHttpSender implements Sender { private final DirectByteSlice bufferView = new DirectByteSlice(); private final long flushIntervalNanos; private final ObjList hosts; + private final HttpTokenProvider httpTokenProvider; private final boolean isTls; private final int maxBackoffMillis; private final int maxNameLength; @@ -94,7 +95,6 @@ public abstract class AbstractLineHttpSender implements Sender { private boolean closed; private int currentAddressIndex; private long flushAfterNanos = Long.MAX_VALUE; - private HttpTokenProvider httpTokenProvider; private boolean isTokenPending; private JsonErrorParser jsonErrorParser; private boolean lastFlushFailed; @@ -159,7 +159,8 @@ protected AbstractLineHttpSender( this(new ObjList<>(host), IntList.createWithValues(port), path, clientConfiguration, tlsConfig, client, autoFlushRows, authToken, username, password, maxNameLength, maxRetriesNanos, maxBackoffMillis, minRequestThroughput, flushIntervalNanos, 0, - rnd + rnd, + null ); } @@ -181,7 +182,8 @@ protected AbstractLineHttpSender( long minRequestThroughput, long flushIntervalNanos, int currentAddressIndex, - Rnd rnd + Rnd rnd, + HttpTokenProvider httpTokenProvider ) { assert authToken == null || (username == null && password == null); this.maxRetriesNanos = maxRetriesNanos; @@ -192,6 +194,7 @@ protected AbstractLineHttpSender( this.path = path != null ? path : PATH; this.autoFlushRows = autoFlushRows; this.authToken = authToken; + this.httpTokenProvider = httpTokenProvider; this.username = username; this.password = password; this.minRequestThroughput = minRequestThroughput; @@ -200,18 +203,25 @@ protected AbstractLineHttpSender( this.isTls = tlsConfig != null; - if (client != null) { - this.client = client; - } else { - this.client = isTls ? - HttpClientFactory.newTlsInstance(clientConfiguration, tlsConfig) - : HttpClientFactory.newPlainTextInstance(clientConfiguration); + // Close the supplied or newly created client if sender initialization fails. + try { + if (client != null) { + this.client = client; + } else { + this.client = isTls ? + HttpClientFactory.newTlsInstance(clientConfiguration, tlsConfig) + : HttpClientFactory.newPlainTextInstance(clientConfiguration); + } + this.questDBVersion = new BuildInformationHolder().getSwVersion(); + // Precompute the User-Agent header value once: newRequest() runs on every flush, so + // concatenating it there would allocate a String each time. + this.userAgent = "QuestDB/java/" + questDBVersion; + this.request = newRequest(); + } catch (Throwable th) { + Misc.freeSuppressing(this.client, th); + this.client = null; + throw th; } - this.questDBVersion = new BuildInformationHolder().getSwVersion(); - // precompute the User-Agent header value once: newRequest() runs on every flush, so concatenating it - // there would allocate a String each time - this.userAgent = "QuestDB/java/" + questDBVersion; - this.request = newRequest(); this.maxNameLength = maxNameLength; this.rnd = rnd; } @@ -406,7 +416,8 @@ public static AbstractLineHttpSender createLineSender( minRequestThroughput, flushIntervalNanos, currentAddressIndex, - rnd + rnd, + httpTokenProvider ); break; case PROTOCOL_VERSION_V2: @@ -427,7 +438,8 @@ public static AbstractLineHttpSender createLineSender( minRequestThroughput, flushIntervalNanos, currentAddressIndex, - rnd + rnd, + httpTokenProvider ); break; case PROTOCOL_VERSION_V3: @@ -448,23 +460,13 @@ public static AbstractLineHttpSender createLineSender( minRequestThroughput, flushIntervalNanos, currentAddressIndex, - rnd + rnd, + httpTokenProvider ); break; default: throw new LineSenderException("Unsupported protocol version: " + protocolVersion); } - if (httpTokenProvider != null) { - // The constructor built the initial request before the provider was wired (httpTokenProvider was - // still null, so it took the no-auth path with withContent). Rebuild it via the deferred path now - // that the provider is set: this leaves the request at the header stage with the token pending, - // matching the reset() path, so the first row's stampTokenIfPending() finishes it (appends the auth - // header + withContent()) without a second client.newRequest(). Deferring the first getToken() off - // the build path also lets a lazily-signing-in provider (e.g. OidcDeviceAuth::getToken) be wired - // before sign-in completes, keeping the token pull on the use/flush path the provider documents. - sender.httpTokenProvider = httpTokenProvider; - sender.request = sender.newRequest(); - } return sender; } diff --git a/core/src/main/java/io/questdb/client/cutlass/line/http/LineHttpSenderV1.java b/core/src/main/java/io/questdb/client/cutlass/line/http/LineHttpSenderV1.java index 9ecba1ca0..4ff11ff57 100644 --- a/core/src/main/java/io/questdb/client/cutlass/line/http/LineHttpSenderV1.java +++ b/core/src/main/java/io/questdb/client/cutlass/line/http/LineHttpSenderV1.java @@ -26,6 +26,7 @@ import io.questdb.client.ClientTlsConfiguration; import io.questdb.client.HttpClientConfiguration; +import io.questdb.client.HttpTokenProvider; import io.questdb.client.Sender; import io.questdb.client.cairo.MicrosTimestampDriver; import io.questdb.client.cairo.NanosTimestampDriver; @@ -94,7 +95,8 @@ protected LineHttpSenderV1(ObjList hosts, long minRequestThroughput, long flushIntervalNanos, int currentAddressIndex, - Rnd rnd) { + Rnd rnd, + HttpTokenProvider httpTokenProvider) { super(hosts, ports, path, @@ -111,7 +113,8 @@ protected LineHttpSenderV1(ObjList hosts, minRequestThroughput, flushIntervalNanos, currentAddressIndex, - rnd); + rnd, + httpTokenProvider); } @Override diff --git a/core/src/main/java/io/questdb/client/cutlass/line/http/LineHttpSenderV2.java b/core/src/main/java/io/questdb/client/cutlass/line/http/LineHttpSenderV2.java index a69b99bfa..0a422840f 100644 --- a/core/src/main/java/io/questdb/client/cutlass/line/http/LineHttpSenderV2.java +++ b/core/src/main/java/io/questdb/client/cutlass/line/http/LineHttpSenderV2.java @@ -26,6 +26,7 @@ import io.questdb.client.ClientTlsConfiguration; import io.questdb.client.HttpClientConfiguration; +import io.questdb.client.HttpTokenProvider; import io.questdb.client.Sender; import io.questdb.client.cairo.ColumnType; import io.questdb.client.cairo.MicrosTimestampDriver; @@ -99,7 +100,8 @@ public LineHttpSenderV2( long minRequestThroughput, long flushIntervalNanos, int currentAddressIndex, - Rnd rnd + Rnd rnd, + HttpTokenProvider httpTokenProvider ) { super( hosts, @@ -118,7 +120,8 @@ public LineHttpSenderV2( minRequestThroughput, flushIntervalNanos, currentAddressIndex, - rnd + rnd, + httpTokenProvider ); } diff --git a/core/src/main/java/io/questdb/client/cutlass/line/http/LineHttpSenderV3.java b/core/src/main/java/io/questdb/client/cutlass/line/http/LineHttpSenderV3.java index eff6acd5f..e911df860 100644 --- a/core/src/main/java/io/questdb/client/cutlass/line/http/LineHttpSenderV3.java +++ b/core/src/main/java/io/questdb/client/cutlass/line/http/LineHttpSenderV3.java @@ -26,6 +26,7 @@ import io.questdb.client.ClientTlsConfiguration; import io.questdb.client.HttpClientConfiguration; +import io.questdb.client.HttpTokenProvider; import io.questdb.client.Sender; import io.questdb.client.cutlass.http.client.HttpClient; import io.questdb.client.cutlass.line.EntityTypes; @@ -59,7 +60,8 @@ public LineHttpSenderV3( long minRequestThroughput, long flushIntervalNanos, int currentAddressIndex, - Rnd rnd + Rnd rnd, + HttpTokenProvider httpTokenProvider ) { super( hosts, @@ -78,7 +80,8 @@ public LineHttpSenderV3( minRequestThroughput, flushIntervalNanos, currentAddressIndex, - rnd + rnd, + httpTokenProvider ); } diff --git a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java index 41cc0a8c9..2f9c5a122 100644 --- a/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java +++ b/core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java @@ -933,11 +933,7 @@ public static QwpWebSocketSender connectWithCredentialSupplier( // the logical slot lock is held one frame up. Closing the engine with the // default reclaim would unlink the lock file build() is still holding. sender.reclaimLogicalSlotLockOnClose = false; - try { - sender.close(); - } catch (Throwable closeFailure) { - t.addSuppressed(closeFailure); - } + Misc.freeSuppressing(sender, t); throw t; } return sender; 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 912d3b444..1a4390269 100644 --- a/core/src/main/java/io/questdb/client/impl/SenderPool.java +++ b/core/src/main/java/io/questdb/client/impl/SenderPool.java @@ -38,6 +38,7 @@ import io.questdb.client.cutlass.qwp.client.sf.cursor.SlotLockContentionException; import io.questdb.client.std.Files; import io.questdb.client.std.IntList; +import io.questdb.client.std.Misc; import org.jetbrains.annotations.TestOnly; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -1972,13 +1973,7 @@ private SenderSlot createSlot(IntFunction factory, int slotIndex) { addSuppressed(failure, deregistrationFailure); } } - if (delegate != null) { - try { - delegate.close(); - } catch (Throwable closeFailure) { - addSuppressed(failure, closeFailure); - } - } + Misc.freeSuppressing(delegate, failure); throw failure; } } diff --git a/core/src/main/java/io/questdb/client/std/Misc.java b/core/src/main/java/io/questdb/client/std/Misc.java index d189a4569..f39f5f88c 100644 --- a/core/src/main/java/io/questdb/client/std/Misc.java +++ b/core/src/main/java/io/questdb/client/std/Misc.java @@ -64,6 +64,23 @@ public static T freeIfCloseable(T object) { return null; } + // Close during rollback, attaching cleanup failures without replacing the original failure. + public static void freeSuppressing(Closeable object, Throwable failure) { + if (object != null) { + try { + object.close(); + } catch (Throwable closeFailure) { + if (closeFailure != failure) { + try { + failure.addSuppressed(closeFailure); + } catch (Throwable ignored) { + // Recording the suppressed failure can itself run out of memory. + } + } + } + } + } + public static Decimal128 getThreadLocalDecimal128() { return tlDecimal128.get(); } 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 6974d0b54..76268a311 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 000000000..c15fb3715 --- /dev/null +++ b/core/src/test/java/io/questdb/client/test/cutlass/http/client/HttpClientConstructorTest.java @@ -0,0 +1,203 @@ +/*+***************************************************************************** + * ___ _ ____ ____ + * / _ \ _ _ ___ ___| |_| _ \| __ ) + * | | | | | | |/ _ \/ __| __| | | | _ \ + * | |_| | |_| | __/\__ \ |_| |_| | |_) | + * \__\_\\__,_|\___||___/\__|____/|____/ + * + * 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.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.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 testBaseConstructorFailurePreservesOriginalFailureWhenSocketCloseFails() throws Exception { + final HttpClientConfiguration configuration = new DefaultHttpClientConfiguration() { + @Override + public int getInitialRequestBufferSize() { + return 1024; + } + + @Override + public int getResponseBufferSize() { + return -1; + } + }; + + TestUtils.assertMemoryLeak(() -> { + final CloseFailure closeFailure = new CloseFailure(); + final SocketFactory socketFactory = (nf, log) -> new PlainSocket(nf, log) { + @Override + public synchronized void close() { + super.close(); + throw closeFailure; + } + }; + try { + buildAndClose(() -> HttpClientFactory.newInstance(configuration, socketFactory)); + Assert.fail("expected IllegalArgumentException"); + } catch (IllegalArgumentException constructionFailure) { + Assert.assertArrayEquals( + "socket close failure must remain secondary", + new Throwable[]{closeFailure}, + constructionFailure.getSuppressed() + ); + } + }); + } + + @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)); + } + + 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 CloseFailure extends RuntimeException { + CloseFailure() { + super("injected socket close failure", null, false, false); + } + } + + 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 000000000..74ea54c8c --- /dev/null +++ b/core/src/test/java/io/questdb/client/test/cutlass/line/LineHttpSenderConstructorTest.java @@ -0,0 +1,202 @@ +/*+***************************************************************************** + * ___ _ ____ ____ + * / _ \ _ _ ___ ___| |_| _ \| __ ) + * | | | | | | |/ _ \/ __| __| | | | _ \ + * | |_| | |_| | __/\__ \ |_| |_| | |_) | + * \__\_\\__,_|\___||___/\__|____/|____/ + * + * 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. + */ +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 testProviderClientIsReleasedWhenRequestPreambleDoesNotFit() throws Exception { + for (int protocolVersion : new int[]{Sender.PROTOCOL_VERSION_V1, Sender.PROTOCOL_VERSION_V2, Sender.PROTOCOL_VERSION_V3}) { + 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, + protocolVersion, + () -> { + Assert.fail("provider must not be queried during construction"); + return null; + } + )); + Assert.fail("expected HttpClientException"); + } catch (HttpClientException ignore) { + // newRequest() could not fit the preamble + } + }); + } + } + + @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(), + null + ); + } + + 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/cutlass/line/LineHttpSenderTokenProviderTest.java b/core/src/test/java/io/questdb/client/test/cutlass/line/LineHttpSenderTokenProviderTest.java index fb94e08e0..773f335e3 100644 --- a/core/src/test/java/io/questdb/client/test/cutlass/line/LineHttpSenderTokenProviderTest.java +++ b/core/src/test/java/io/questdb/client/test/cutlass/line/LineHttpSenderTokenProviderTest.java @@ -418,28 +418,30 @@ public void testNullOrEmptyProviderTokenIsRejected() throws Exception { @Test public void testProviderTokenNotPulledAtBuildAndPulledOnFirstRow() throws Exception { - assertMemoryLeak(() -> { - AtomicInteger calls = new AtomicInteger(); - HttpTokenProvider provider = () -> { - calls.incrementAndGet(); - return "TOKEN"; - }; - try (Sender sender = Sender.builder(Sender.Transport.HTTP) - .address("127.0.0.1:1") - .protocolVersion(Sender.PROTOCOL_VERSION_V1) - .disableAutoFlush() - .httpTokenProvider(provider) - .build()) { - // build() must not query the provider: a lazily-signing-in provider would not have a token yet - Assert.assertEquals("provider must not be queried at build time", 0, calls.get()); - // the first row pulls the deferred token so the first send will carry it - sender.table("t").longColumn("v", 1L).atNow(); - Assert.assertEquals("provider must be queried when the first row starts", 1, calls.get()); - // a second row in the same un-flushed batch reuses the same request, so it does not re-pull - sender.table("t").longColumn("v", 2L).atNow(); - Assert.assertEquals("provider must not be re-queried within the same batch", 1, calls.get()); - } - }); + for (int protocolVersion : new int[]{Sender.PROTOCOL_VERSION_V1, Sender.PROTOCOL_VERSION_V2, Sender.PROTOCOL_VERSION_V3}) { + assertMemoryLeak(() -> { + AtomicInteger calls = new AtomicInteger(); + HttpTokenProvider provider = () -> { + calls.incrementAndGet(); + return "TOKEN"; + }; + try (Sender sender = Sender.builder(Sender.Transport.HTTP) + .address("127.0.0.1:1") + .protocolVersion(protocolVersion) + .disableAutoFlush() + .httpTokenProvider(provider) + .build()) { + // build() must not query the provider: a lazily-signing-in provider would not have a token yet + Assert.assertEquals("provider must not be queried at build time", 0, calls.get()); + // the first row pulls the deferred token so the first send will carry it + sender.table("t").longColumn("v", 1L).atNow(); + Assert.assertEquals("provider must be queried when the first row starts", 1, calls.get()); + // a second row in the same un-flushed batch reuses the same request, so it does not re-pull + sender.table("t").longColumn("v", 2L).atNow(); + Assert.assertEquals("provider must not be re-queried within the same batch", 1, calls.get()); + } + }); + } } @Test(timeout = 30_000) diff --git a/core/src/test/java/io/questdb/client/test/std/MiscTest.java b/core/src/test/java/io/questdb/client/test/std/MiscTest.java new file mode 100644 index 000000000..addc7f4c7 --- /dev/null +++ b/core/src/test/java/io/questdb/client/test/std/MiscTest.java @@ -0,0 +1,79 @@ +/*+***************************************************************************** + * ___ _ ____ ____ + * / _ \ _ _ ___ ___| |_| _ \| __ ) + * | | | | | | |/ _ \/ __| __| | | | _ \ + * | |_| | |_| | __/\__ \ |_| |_| | |_) | + * \__\_\\__,_|\___||___/\__|____/|____/ + * + * 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.std; + +import io.questdb.client.std.Misc; +import org.junit.Assert; +import org.junit.Test; + +import java.io.IOException; +import java.util.concurrent.atomic.AtomicInteger; + +public class MiscTest { + @Test + public void testFreeSuppressingClosesResource() { + Throwable failure = new IllegalStateException("construction failed"); + AtomicInteger closeCalls = new AtomicInteger(); + Misc.freeSuppressing(closeCalls::incrementAndGet, failure); + Assert.assertEquals(1, closeCalls.get()); + Assert.assertEquals(0, failure.getSuppressed().length); + } + + @Test + public void testFreeSuppressingHandlesNull() { + Throwable failure = new IllegalStateException("construction failed"); + Misc.freeSuppressing(null, failure); + Assert.assertEquals(0, failure.getSuppressed().length); + } + + @Test + public void testFreeSuppressingSkipsSelfSuppression() { + AssertionError failure = new AssertionError("construction and close failed"); + Misc.freeSuppressing(() -> { + throw failure; + }, failure); + Assert.assertEquals(0, failure.getSuppressed().length); + } + + @Test + public void testFreeSuppressingSuppressesError() { + Throwable failure = new IllegalStateException("construction failed"); + AssertionError closeFailure = new AssertionError("close failed"); + Misc.freeSuppressing(() -> { + throw closeFailure; + }, failure); + Assert.assertArrayEquals(new Throwable[]{closeFailure}, failure.getSuppressed()); + } + + @Test + public void testFreeSuppressingSuppressesIOException() { + Throwable failure = new IllegalStateException("construction failed"); + IOException closeFailure = new IOException("close failed"); + Misc.freeSuppressing(() -> { + throw closeFailure; + }, failure); + Assert.assertArrayEquals(new Throwable[]{closeFailure}, failure.getSuppressed()); + } +}