From a8fcec78861f8ef9d999c9fd855f9c64af0d6b1c Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 19 Aug 2026 13:56:31 +0300 Subject: [PATCH 1/4] [#872] Bound the connect of the JDBC pool and report a connect it cannot make CachedConnection.getConnection() established connections with no bound of its own and treated every SQLException from the connect as "the server is at max_connections", retrying it recursively with a wait doubling from 1 ms and no end to it. A database that listens but does not answer, a password that is not accepted, a driver that is not in lib/extensions - each hung the caller instead of failing it, silently: every backend operation, the open of a backend and dsconfig create-backend-index on a running server included, borrows through this path. The borrow is now bounded in both phases. One connect attempt is bounded by the properties of its dialect, recognized by the prefix of the connection string, through org.openidentityplatform.opendj.jdbc.connect.timeout (30 s by default, 0 for no bound); a property the connection string sets itself keeps precedence, so the loginTimeout/socketTimeout an administrator put into db-directory by hand still governs. Not one of the four drivers bounds the attempt with a single property - the second covers the reads of the prelogin handshake, of TLS and of authentication - and that includes the SQL Server driver, whose loginTimeout leaves the prelogin read open. Where that second property is a socket read timeout for the life of the connection (mysql, oracle, sql server), it is lifted once the login is through, so a statement slower than the bound is unaffected. Only a database that accepts no further connection is retried now, under the deadline of org.openidentityplatform.opendj.jdbc.pool.timeout (60 s by default), with the backoff capped at 1 s and a throttled warning so the stall is visible in the server log; every other failure is reported to the caller. A connect whose setup fails no longer leaks the connection, a connection that cannot be rolled back is closed instead of being pooled or dropped, and a pooled connection is validated with a bound rather than with isValid(0), which means "no timeout" in the JDBC contract. CachedConnectionTestCase covers all of it without a database - every dialect against a socket that never answers and a driver of the test for the retry - and the container suites assert that the read bound of the login does not outlive it. --- .../backends/jdbc/CachedConnection.java | 344 ++++++++++++-- .../jdbc/CachedConnectionTestCase.java | 443 ++++++++++++++++++ .../opends/server/backends/jdbc/TestCase.java | 21 + 3 files changed, 778 insertions(+), 30 deletions(-) create mode 100644 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java index 188f7ba2fb..874b8162fe 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java @@ -23,9 +23,12 @@ import java.sql.*; import java.time.Duration; +import java.util.Locale; import java.util.Map; import java.util.Properties; import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; public class CachedConnection implements Connection { private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass(); @@ -33,6 +36,28 @@ public class CachedConnection implements Connection { static final String TTL_PROPERTY = "org.openidentityplatform.opendj.jdbc.ttl"; static final long DEFAULT_TTL_MS = 15000; + /** Bounds the connect and the login of one attempt to establish a connection, in seconds; 0 for no bound. */ + static final String CONNECT_TIMEOUT_PROPERTY = "org.openidentityplatform.opendj.jdbc.connect.timeout"; + static final long DEFAULT_CONNECT_TIMEOUT_SECONDS = 30; + + /** Bounds a whole borrow - every connect attempt and every wait for a pooled connection - in seconds; 0 for no bound. */ + static final String POOL_TIMEOUT_PROPERTY = "org.openidentityplatform.opendj.jdbc.pool.timeout"; + static final long DEFAULT_POOL_TIMEOUT_SECONDS = 60; + + /** Bound of the validation of a pooled connection: isValid(0) means "no timeout" in the JDBC contract. */ + static final int VALIDATION_TIMEOUT_SECONDS = 5; + + static final long MAX_BACKOFF_MS = 1000; + static final long STALL_WARNING_AFTER_MS = 1000; + static final long STALL_WARNING_INTERVAL_MS = 10000; + + // setNetworkTimeout() takes the executor its timeout handling runs on: the drivers it is used + // with here only set a socket option in it, so it costs a call rather than a thread. + private static final Executor DIRECT_EXECUTOR = Runnable::run; + + private static final AtomicLong lastStallWarning = new AtomicLong(); + private static final AtomicBoolean readBoundWarned = new AtomicBoolean(); + final Connection parent; static LoadingCache> cached = Caffeine.newBuilder() @@ -55,19 +80,134 @@ public class CachedConnection implements Connection { * {@value #TTL_PROPERTY} system property. An invalid value is ignored in favor of the default. */ private static long getCacheTtlMillis() { - final String ttl = System.getProperty(TTL_PROPERTY); - if (ttl != null) { + return getNonNegativeProperty(TTL_PROPERTY, DEFAULT_TTL_MS); + } + + /** + * Returns the value of a numeric system property, ignoring a value that is not a non-negative + * number in favor of the default. + */ + private static long getNonNegativeProperty(String name, long defaultValue) { + final String value = System.getProperty(name); + if (value != null) { try { - final long millis = Long.parseLong(ttl.trim()); - if (millis >= 0) { - return millis; + final long parsed = Long.parseLong(value.trim()); + if (parsed >= 0) { + return parsed; } } catch (NumberFormatException ignored) { } - logger.warn(LocalizableMessage.raw("Ignoring invalid value \"%s\" of the %s property, using %d ms", - ttl, TTL_PROPERTY, DEFAULT_TTL_MS)); + logger.warn(LocalizableMessage.raw("Ignoring invalid value \"%s\" of the %s property, using %d", + value, name, defaultValue)); + } + return defaultValue; + } + + /** + * The drivers this backend is used with, recognized by the prefix of the connection string, + * together with the properties that bound one attempt to establish a connection. Not one of + * them bounds the attempt with a single property: the one named first covers the socket + * connect, and the login behind it - the reads of the prelogin handshake, of TLS and of + * authentication, the phase a proxy at its connection limit or a moved VIP leaves unanswered - + * needs the second. That holds for the SQL Server driver too, whose loginTimeout leaves the + * read of the prelogin answer unbounded (CachedConnectionTestCase covers every one of them + * against a socket that never answers). + */ + enum ConnectDialect { + /** postgresql: both properties take seconds; loginTimeout bounds the login the driver runs on a thread of its own. */ + POSTGRES("jdbc:postgresql:", "connectTimeout", 1, "loginTimeout", 1, false, new int[]{}), + /** mysql: both properties take milliseconds; socketTimeout is a socket read timeout that outlives the login. */ + MYSQL("jdbc:mysql:", "connectTimeout", 1000, "socketTimeout", 1000, true, new int[]{1040, 1203}), + /** oracle: both properties take milliseconds; ReadTimeout is a socket read timeout that outlives the login. */ + ORACLE("jdbc:oracle:", "oracle.net.CONNECT_TIMEOUT", 1000, "oracle.jdbc.ReadTimeout", 1000, true, + new int[]{20, 12516, 12518, 12519, 12520}), + /** ms sql server: loginTimeout takes seconds, socketTimeout milliseconds; the latter is a socket read timeout that outlives the login. */ + MICROSOFT("jdbc:sqlserver:", "loginTimeout", 1, "socketTimeout", 1000, true, new int[]{17809, 10928, 10929}); + + final String urlPrefix; + final String connectProperty; + final int connectUnitsPerSecond; + final String readProperty; + final int readUnitsPerSecond; + /** whether the read bound of the login stays in force for every statement issued afterwards */ + final boolean readBoundOutlivesLogin; + /** the vendor codes of this dialect for "no further connection is accepted" */ + final int[] connectionLimitCodes; + + ConnectDialect(String urlPrefix, String connectProperty, int connectUnitsPerSecond, + String readProperty, int readUnitsPerSecond, boolean readBoundOutlivesLogin, + int[] connectionLimitCodes) { + this.urlPrefix = urlPrefix; + this.connectProperty = connectProperty; + this.connectUnitsPerSecond = connectUnitsPerSecond; + this.readProperty = readProperty; + this.readUnitsPerSecond = readUnitsPerSecond; + this.readBoundOutlivesLogin = readBoundOutlivesLogin; + this.connectionLimitCodes = connectionLimitCodes; + } + + /** The dialect of a connection string, or null for a driver whose property names are not known here. */ + static ConnectDialect of(String connectionString) { + final String url = connectionString.toLowerCase(Locale.ROOT); + for (final ConnectDialect dialect : values()) { + if (url.startsWith(dialect.urlPrefix)) { + return dialect; + } + } + return null; + } + + /** + * Fills in the properties bounding one connect attempt, leaving out every property the + * connection string sets itself - an explicit setting of the administrator keeps + * precedence, and the SQL Server driver gives a supplied property precedence over the url. + * Returns whether a read bound outliving the login was set and has to be lifted once the + * connection is established. + */ + boolean bound(String connectionString, Properties properties, long timeoutSeconds) { + if (!declaredInUrl(connectionString, connectProperty)) { + properties.setProperty(connectProperty, Long.toString(timeoutSeconds * connectUnitsPerSecond)); + } + if (readProperty != null && !declaredInUrl(connectionString, readProperty)) { + properties.setProperty(readProperty, Long.toString(timeoutSeconds * readUnitsPerSecond)); + return readBoundOutlivesLogin; + } + return false; + } + + boolean isConnectionLimit(SQLException e) { + for (final int code : connectionLimitCodes) { + if (e.getErrorCode() == code) { + return true; + } + } + return false; + } + + // Whether the connection string sets this property itself. The dialects separate their + // parameters differently - "?a=1&b=2" (postgresql, mysql), ";a=1;b=2" (sql server), + // "(A=1)" inside the descriptor of an oracle tns url, where the property also goes by the + // last segment of its name alone - so a parameter is recognized by the delimiter in front + // of it and the "=" behind it rather than by parsing the url syntax of every driver. + private static boolean declaredInUrl(String connectionString, String property) { + if (containsParameter(connectionString, property)) { + return true; + } + final int dot = property.lastIndexOf('.'); + return dot >= 0 && containsParameter(connectionString, property.substring(dot + 1)); + } + + private static boolean containsParameter(String connectionString, String property) { + final String url = connectionString.toLowerCase(Locale.ROOT); + final String name = property.toLowerCase(Locale.ROOT); + for (int i = url.indexOf(name); i >= 0; i = url.indexOf(name, i + name.length())) { + final int end = i + name.length(); + if (i > 0 && "?&;(,".indexOf(url.charAt(i - 1)) >= 0 && end < url.length() && url.charAt(end) == '=') { + return true; + } + } + return false; } - return DEFAULT_TTL_MS; } final String connectionString; @@ -76,38 +216,175 @@ public CachedConnection(String connectionString, Connection parent) { this.parent = parent; } + /** + * Borrows a connection: a usable one out of the pool, or a newly established one. Bounded in + * both phases - every operation of this backend, the open of a backend and the import + * included, comes through here, and an unbounded borrow turns a database that listens but does + * not answer into a hang rather than into an error the caller can report. + */ static Connection getConnection(String connectionString) throws Exception { - return getConnection(connectionString, 0); + final ConnectDialect dialect = ConnectDialect.of(connectionString); + final long connectTimeoutSeconds = Math.min( + getNonNegativeProperty(CONNECT_TIMEOUT_PROPERTY, DEFAULT_CONNECT_TIMEOUT_SECONDS), Integer.MAX_VALUE / 1000); + final long poolTimeoutSeconds = getNonNegativeProperty(POOL_TIMEOUT_PROPERTY, DEFAULT_POOL_TIMEOUT_SECONDS); + final long startedAt = System.currentTimeMillis(); + final long deadline = (poolTimeoutSeconds == 0 || poolTimeoutSeconds >= Long.MAX_VALUE / 1000) + ? Long.MAX_VALUE : startedAt + poolTimeoutSeconds * 1000; + long waitMs = 0; + long backoffMs = 0; + int attempts = 0; + while (true) { + final CachedConnection pooled = poll(connectionString, waitMs); + if (pooled != null) { + return pooled; + } + attempts++; + try { + return connect(connectionString, dialect, connectTimeoutSeconds); + } catch (SQLException e) { + // A database that accepts no further connection is the one failure worth waiting + // out: one of ours is going to come back to the pool. Everything else - a password + // that is not accepted, a database that is down, a driver that is not on the + // classpath - is reported to the caller instead of being retried behind its back. + if (!isConnectionLimit(e, dialect)) { + throw e; + } + final long remaining = deadline - System.currentTimeMillis(); + if (remaining <= 0) { + throw new SQLTimeoutException("no connection to " + safeUrl(connectionString) + " could be borrowed within " + + poolTimeoutSeconds + "s (" + attempts + " attempts): the database accepts no further connection" + + " and none was returned to the pool", e); + } + backoffMs = Math.min(backoffMs == 0 ? 1 : backoffMs * 2, MAX_BACKOFF_MS); + waitMs = Math.min(backoffMs, remaining); + warnStall(connectionString, attempts, startedAt, e); + } + } } - static Connection getConnection(String connectionString, final int waitTime) throws Exception { - CachedConnection con = cached.get(connectionString).poll(waitTime, TimeUnit.MILLISECONDS); - + /** Takes a usable connection out of the pool, waiting up to waitMs for one to be returned to it. */ + private static CachedConnection poll(String connectionString, long waitMs) throws InterruptedException { + CachedConnection con = cached.get(connectionString).poll(waitMs, TimeUnit.MILLISECONDS); while (con != null) { - if (!con.isValid(0)) { - try { - con.parent.close(); - } catch (SQLException e) { - con = null; - } - con = cached.get(connectionString).poll(); - } else { + if (isUsable(con)) { return con; } + closeQuietly(con.parent); + con = cached.get(connectionString).poll(); + } + return null; + } + + private static boolean isUsable(CachedConnection con) { + try { + // The validation needs a bound of its own: isValid(0) means "no timeout" in the JDBC + // contract, and a connection whose socket is half-open answers it no sooner than it + // answers anything else. + return con.isValid(VALIDATION_TIMEOUT_SECONDS); + } catch (SQLException e) { // a driver reporting the validation as an error: discard it + return false; } - Connection conNew = null; + } + + private static CachedConnection connect(String connectionString, ConnectDialect dialect, long connectTimeoutSeconds) + throws SQLException { + // A driver is free to write into the map it is handed, so it gets one of its own. + final Properties properties = new Properties(); + final boolean readBoundSet = dialect != null && connectTimeoutSeconds > 0 + && dialect.bound(connectionString, properties, connectTimeoutSeconds); + final Connection conNew = DriverManager.getConnection(connectionString, properties); try { - conNew = DriverManager.getConnection(connectionString); + // still under the read bound: both of these are round trips of their own conNew.setAutoCommit(false); conNew.setTransactionIsolation(TRANSACTION_READ_COMMITTED); - return new CachedConnection(connectionString, conNew); - } catch (SQLException e) { // max_connection server error: try recursion for reuse connection - if (conNew != null) { // the connection was established but not set up: nothing else would close it - try { - conNew.close(); - } catch (SQLException e2) {} + if (readBoundSet) { + relaxReadBound(conNew); + } + } catch (SQLException e) { // nothing holds this connection yet: it would leak + closeQuietly(conNew); + throw e; + } + return new CachedConnection(connectionString, conNew); + } + + // The second bound of the login is a socket read timeout on mysql, oracle and sql server, in + // force for the whole life of the connection: left in place it would break every statement + // slower than it - an import batch, the statistics of a freshly loaded table - so it is lifted + // as soon as the login is through, restoring the behaviour of a connection this class + // established before. A read bound the connection string sets itself is never touched here: + // it is not set at all, so nothing of the administrator's is lifted along with it. + private static void relaxReadBound(Connection con) { + try { + con.setNetworkTimeout(DIRECT_EXECUTOR, 0); + } catch (SQLException | RuntimeException e) { + if (readBoundWarned.compareAndSet(false, true)) { + logger.warn(LocalizableMessage.raw( + "The read bound of the login could not be lifted (%s): statements taking longer than the %s" + + " property will fail on connections of this backend", e.getMessage(), CONNECT_TIMEOUT_PROPERTY)); } - return getConnection(connectionString, (waitTime == 0) ? 1 : waitTime * 2); + } + } + + /** Whether the database refused the connection because it accepts no further one. */ + static boolean isConnectionLimit(SQLException e, ConnectDialect dialect) { + for (Throwable t = e; t != null; t = t.getCause()) { + if (t instanceof SQLException) { + final SQLException sql = (SQLException) t; + // class 53, insufficient_resources, is how the standard - and postgresql, with + // 53300 too_many_connections - reports a server taking no further connection + final String sqlState = sql.getSQLState(); + if ((sqlState != null && sqlState.startsWith("53")) + || (dialect != null && dialect.isConnectionLimit(sql))) { + return true; + } + } + } + return false; + } + + // A stall has to reach the server log: without it a database accepting no further connection + // is indistinguishable from a hang. Throttled, since every operation of the backend borrows + // through here and would otherwise log a copy of its own. + private static void warnStall(String connectionString, int attempts, long startedAt, SQLException cause) { + final long now = System.currentTimeMillis(); + if (now - startedAt < STALL_WARNING_AFTER_MS) { + return; + } + final long last = lastStallWarning.get(); + if (now - last >= STALL_WARNING_INTERVAL_MS && lastStallWarning.compareAndSet(last, now)) { + logger.warn(LocalizableMessage.raw( + "%s accepts no further connection: waiting %d ms for a pooled one so far (%d attempts), last error: %s", + safeUrl(connectionString), now - startedAt, attempts, cause.getMessage())); + } + } + + // The connection string carries the credentials of the backend, so it is never logged as it + // stands. Parameters - "?user=...&password=..." on postgresql and mysql, ";password=..." on + // sql server - are cut off behind their first separator, while the credentials of a url that + // carries them in front of an "@" ("user/password@//host" on an oracle thin url, the userinfo + // of a url) are cut off in front of it, leaving the scheme and the host they stand between. + static String safeUrl(String connectionString) { + int cut = connectionString.length(); + for (final char separator : new char[]{'?', ';'}) { + final int at = connectionString.indexOf(separator); + if (at >= 0 && at < cut) { + cut = at; + } + } + final String url = connectionString.substring(0, cut); + final int at = url.indexOf('@'); + if (at < 0) { + return url; + } + final int scheme = url.indexOf(':', "jdbc:".length()) + 1; // the end of "jdbc::" + return url.substring(0, scheme) + url.substring(at); + } + + private static void closeQuietly(Connection con) { + try { + con.close(); + } catch (SQLException e) { + // ignore: it is on its way out anyway } } @@ -153,7 +430,14 @@ public void rollback() throws SQLException { @Override public void close() throws SQLException { - rollback(); + try { + rollback(); + } catch (SQLException e) { + // A connection that cannot be rolled back must not be handed to the next borrower - + // and must not be dropped on the floor either: nothing else holds it any more. + closeQuietly(parent); + throw e; + } cached.get(connectionString).add(this); } diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java new file mode 100644 index 0000000000..1590536363 --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java @@ -0,0 +1,443 @@ +/* + * The contents of this file are subject to the terms of the Common Development and + * Distribution License (the License). You may not use this file except in compliance with the + * License. + * + * You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the + * specific language governing permission and limitations under the License. + * + * When distributing Covered Software, include this CDDL Header Notice in each file and include + * the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL + * Header, with the fields enclosed by brackets [] replaced by your own identifying + * information: "Portions Copyright [year] [name of copyright owner]". + * + * Copyright 2026 3A Systems, LLC. + */ +package org.opends.server.backends.jdbc; + +import org.opends.server.DirectoryServerTestCase; +import org.testng.annotations.AfterClass; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import java.net.InetAddress; +import java.net.ServerSocket; +import java.sql.Connection; +import java.sql.Driver; +import java.sql.DriverManager; +import java.sql.DriverPropertyInfo; +import java.sql.SQLException; +import java.sql.SQLTimeoutException; +import java.util.Properties; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.FutureTask; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.logging.Logger; + +import static org.mockito.Mockito.anyInt; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertSame; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.fail; + +/** + * The pool every operation of the JDBC backend borrows from must bound both of its phases and + * report a connect it cannot make, rather than retrying it out of sight of the caller (#872). + * Needs no database: the dialects are exercised against a socket that never answers and against a + * driver of this test, so a regression fails the build wherever it runs. + */ +@SuppressWarnings("javadoc") +@Test(groups = { "precommit", "jdbc" }, sequential = true) +public class CachedConnectionTestCase extends DirectoryServerTestCase { + + /** A connect attempt of a bounded dialect must give up in about this long, plus room for a slow machine. */ + private static final long BOUND_SECONDS = 2; + private static final long BOUND_MARGIN_MS = 60000; + + private final StubDriver stub = new StubDriver(); + + @BeforeClass + public void registerStubDriver() throws Exception { + DriverManager.registerDriver(stub); + } + + @AfterClass + public void deregisterStubDriver() throws Exception { + DriverManager.deregisterDriver(stub); + } + + @AfterMethod + public void clearProperties() { + System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY); + System.clearProperty(CachedConnection.POOL_TIMEOUT_PROPERTY); + } + + /** + * A driver that is not on the classpath - the JDBC backend needs one dropped into + * lib/extensions by hand - is a configuration error the caller has to see. Retried, it is + * indistinguishable from a database that hangs. + */ + @Test(timeOut = 120000) + public void testMissingDriverIsReportedAtOnce() throws Exception { + final long startedAt = System.currentTimeMillis(); + try { + CachedConnection.getConnection("jdbc:nosuchengine://127.0.0.1:5432/opendj"); + fail("a connection string no registered driver accepts must be reported"); + } catch (SQLException expected) { + assertTrue(expected.getMessage().contains("No suitable driver"), expected.getMessage()); + } + assertElapsedWithinBound(startedAt, 0); + } + + /** + * A database that is not listening at all: every dialect reports it instead of retrying the + * refused connect until the caller gives up on the operation. + */ + @Test(timeOut = 120000) + public void testRefusedConnectIsReportedAtOnce() throws Exception { + final int closedPort = closedPort(); + System.setProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY, Long.toString(BOUND_SECONDS)); + for (final String url : urlsOf(closedPort)) { + final long startedAt = System.currentTimeMillis(); + try { + CachedConnection.getConnection(url); + fail("a refused connect must be reported: " + CachedConnection.safeUrl(url)); + } catch (SQLException expected) { + // the failure of the moment, reported rather than retried + } + assertElapsedWithinBound(startedAt, BOUND_SECONDS * 1000); + } + } + + /** + * The failure this bound exists for: a database that completes the TCP connection and then + * says nothing - a moved VIP, a proxy at its connection limit, a host that lost its answer - + * leaving the login of the driver, and with it the operation, without an end. The connection + * of the accept queue is never answered here, so every dialect has to give up on its own. + */ + @Test(timeOut = 300000) + public void testLoginIsBoundedWhenTheDatabaseNeverAnswers() throws Exception { + System.setProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY, Long.toString(BOUND_SECONDS)); + // a socket that is bound and never accepted: the kernel completes the handshake, so the + // connect of the driver succeeds and every read of the login that follows hangs + try (final ServerSocket blackhole = new ServerSocket(0, 50, InetAddress.getLoopbackAddress())) { + for (final String url : urlsOf(blackhole.getLocalPort())) { + // borrowed on a thread of its own: a bound that a driver does not honour has to + // fail this test at once, and not by hanging the run it is part of + final FutureTask borrow = new FutureTask<>(() -> CachedConnection.getConnection(url)); + final Thread thread = new Thread(borrow, "borrow-" + CachedConnection.safeUrl(url)); + thread.setDaemon(true); + thread.start(); + try { + final Connection con = borrow.get(BOUND_SECONDS * 1000 + BOUND_MARGIN_MS, TimeUnit.MILLISECONDS); + fail("a database that never answers must not hand out a connection: " + con); + } catch (TimeoutException e) { + fail("the login of " + CachedConnection.safeUrl(url) + " is not bounded: it never gave up"); + } catch (ExecutionException expected) { + assertTrue(expected.getCause() instanceof SQLException, String.valueOf(expected.getCause())); + } + } + } + } + + /** Pool exhaustion stays a retry - one of our own connections is on its way back to the pool. */ + @Test(timeOut = 120000) + public void testConnectionLimitIsRetried() throws Exception { + final String url = StubDriver.PREFIX + "retried"; + stub.failWith(tooManyConnections(), 2); + System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "30"); + + final Connection con = CachedConnection.getConnection(url); + + assertNotNull(con); + assertEquals(stub.attempts.get(), 3, "the connect must be retried while the database is at its limit"); + } + + /** ... but under a deadline: the retry used to double its wait from 1 ms with no end to it. */ + @Test(timeOut = 120000) + public void testConnectionLimitGivesUpAtTheDeadline() throws Exception { + final String url = StubDriver.PREFIX + "deadline"; + stub.failWith(tooManyConnections(), StubDriver.ALWAYS); + System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "2"); + + final long startedAt = System.currentTimeMillis(); + try { + CachedConnection.getConnection(url); + fail("a database that stays at its connection limit must be reported, not waited out forever"); + } catch (SQLTimeoutException expected) { + assertTrue(expected.getMessage().contains("2s"), expected.getMessage()); + assertEquals(((SQLException) expected.getCause()).getSQLState(), "53300"); + } + final long elapsed = System.currentTimeMillis() - startedAt; + assertTrue(elapsed >= 2000, "gave up after " + elapsed + " ms, before the deadline it was given"); + assertElapsedWithinBound(startedAt, 2000); + assertTrue(stub.attempts.get() > 1, "the connect must be retried while the deadline lasts"); + } + + /** + * Every other failure is the caller's to report. A password the database does not accept is + * never going to be accepted by waiting, and the retry that swallowed it left the operation + * hanging with nothing in the log. + */ + @Test(timeOut = 120000) + public void testRejectedLoginIsNotRetried() throws Exception { + final String url = StubDriver.PREFIX + "rejected"; + stub.failWith(new SQLException("password authentication failed", "28P01"), StubDriver.ALWAYS); + System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "30"); + + try { + CachedConnection.getConnection(url); + fail("a rejected login must be reported to the caller"); + } catch (SQLException expected) { + assertEquals(expected.getSQLState(), "28P01"); + } + assertEquals(stub.attempts.get(), 1, "a rejected login must be attempted once"); + } + + /** A connection the setup of which failed belongs to nobody: it has to be closed, not leaked. */ + @Test(timeOut = 120000) + public void testConnectionIsClosedWhenItsSetupFails() throws Exception { + final String url = StubDriver.PREFIX + "setup-failure"; + final Connection broken = mock(Connection.class); + doThrow(new SQLException("read only")).when(broken).setAutoCommit(false); + stub.answerWith(broken); + + try { + CachedConnection.getConnection(url); + fail("a connection that cannot be set up must be reported"); + } catch (SQLException expected) { + assertEquals(expected.getMessage(), "read only"); + } + verify(broken).close(); + } + + /** A pooled connection that no longer validates is closed and replaced, not handed out. */ + @Test(timeOut = 120000) + public void testBrokenPooledConnectionIsDiscarded() throws Exception { + final String url = StubDriver.PREFIX + "broken-pooled"; + final Connection stale = mock(Connection.class); + when(stale.isValid(anyInt())).thenReturn(false); + CachedConnection.cached.get(url).add(new CachedConnection(url, stale)); + final Connection fresh = mock(Connection.class); + when(fresh.isValid(anyInt())).thenReturn(true); + stub.answerWith(fresh); + + final Connection borrowed = CachedConnection.getConnection(url); + + assertSame(((CachedConnection) borrowed).parent, fresh); + verify(stale).close(); + // the validation of a pooled connection needs a bound of its own as well + verify(stale, never()).isValid(0); + verify(stale).isValid(CachedConnection.VALIDATION_TIMEOUT_SECONDS); + } + + /** A connection that cannot be rolled back must not go back into the pool - nor be dropped. */ + @Test(timeOut = 120000) + public void testConnectionThatCannotBeRolledBackIsClosed() throws Exception { + final String url = StubDriver.PREFIX + "rollback-failure"; + final Connection parent = mock(Connection.class); + doThrow(new SQLException("connection is closed")).when(parent).rollback(); + + try { + new CachedConnection(url, parent).close(); + fail("a failed rollback must be reported"); + } catch (SQLException expected) { + assertEquals(expected.getMessage(), "connection is closed"); + } + verify(parent).close(); + assertTrue(CachedConnection.cached.get(url).isEmpty(), "a connection that cannot be rolled back was pooled"); + } + + @Test + public void testDialectIsRecognizedByTheConnectionString() throws Exception { + assertEquals(CachedConnection.ConnectDialect.of("jdbc:postgresql://h:5432/db"), CachedConnection.ConnectDialect.POSTGRES); + assertEquals(CachedConnection.ConnectDialect.of("jdbc:mysql://h:3306/db"), CachedConnection.ConnectDialect.MYSQL); + assertEquals(CachedConnection.ConnectDialect.of("jdbc:oracle:thin:@//h:1521/svc"), CachedConnection.ConnectDialect.ORACLE); + assertEquals(CachedConnection.ConnectDialect.of("jdbc:sqlserver://h:1433;databaseName=db"), CachedConnection.ConnectDialect.MICROSOFT); + assertNull(CachedConnection.ConnectDialect.of("jdbc:h2:mem:db"), "an unknown engine must not be fed the properties of another"); + } + + /** Both phases are bounded, in the units of the driver: the connect alone leaves the login open. */ + @Test + public void testBothPhasesOfTheLoginAreBounded() throws Exception { + final Properties postgres = new Properties(); + assertFalse(CachedConnection.ConnectDialect.POSTGRES.bound("jdbc:postgresql://h:5432/db", postgres, 7)); + assertEquals(postgres.getProperty("connectTimeout"), "7"); + assertEquals(postgres.getProperty("loginTimeout"), "7"); + + final Properties mysql = new Properties(); + assertTrue(CachedConnection.ConnectDialect.MYSQL.bound("jdbc:mysql://h:3306/db", mysql, 7), + "the read bound of mysql outlives the login and has to be lifted"); + assertEquals(mysql.getProperty("connectTimeout"), "7000"); + assertEquals(mysql.getProperty("socketTimeout"), "7000"); + + final Properties oracle = new Properties(); + assertTrue(CachedConnection.ConnectDialect.ORACLE.bound("jdbc:oracle:thin:@//h:1521/svc", oracle, 7)); + assertEquals(oracle.getProperty("oracle.net.CONNECT_TIMEOUT"), "7000"); + assertEquals(oracle.getProperty("oracle.jdbc.ReadTimeout"), "7000"); + + // the loginTimeout of the sql server driver leaves the read of the prelogin answer open + final Properties microsoft = new Properties(); + assertTrue(CachedConnection.ConnectDialect.MICROSOFT.bound("jdbc:sqlserver://h:1433;databaseName=db", microsoft, 7)); + assertEquals(microsoft.getProperty("loginTimeout"), "7"); + assertEquals(microsoft.getProperty("socketTimeout"), "7000"); + } + + /** + * A bound the administrator put into the connection string by hand - the only workaround this + * backend had - keeps precedence, property by property. + */ + @Test + public void testConnectionStringKeepsPrecedence() throws Exception { + final Properties postgres = new Properties(); + CachedConnection.ConnectDialect.POSTGRES.bound( + "jdbc:postgresql://h:5432/db?user=u&password=p&loginTimeout=30&socketTimeout=300", postgres, 7); + assertNull(postgres.getProperty("loginTimeout"), "the setting of the connection string was overridden"); + assertEquals(postgres.getProperty("connectTimeout"), "7", "the property it leaves open must still be bounded"); + + // the sql server driver gives a supplied property precedence over the one of the url + final Properties microsoft = new Properties(); + CachedConnection.ConnectDialect.MICROSOFT.bound("jdbc:sqlserver://h:1433;loginTimeout=45;databaseName=db", microsoft, 7); + assertNull(microsoft.getProperty("loginTimeout"), "the setting of the connection string was overridden"); + assertEquals(microsoft.getProperty("socketTimeout"), "7000"); + + // inside the descriptor of an oracle tns url the property goes by the last segment of its name + final Properties oracle = new Properties(); + CachedConnection.ConnectDialect.ORACLE.bound( + "jdbc:oracle:thin:@(DESCRIPTION=(CONNECT_TIMEOUT=3)(ADDRESS=(HOST=h)(PORT=1521)))", oracle, 7); + assertNull(oracle.getProperty("oracle.net.CONNECT_TIMEOUT")); + assertEquals(oracle.getProperty("oracle.jdbc.ReadTimeout"), "7000"); + + // a name that only appears as the tail of another parameter is not a setting of its own + final Properties mysql = new Properties(); + CachedConnection.ConnectDialect.MYSQL.bound("jdbc:mysql://h:3306/db?xconnectTimeout=1&socketTimeoutX=2", mysql, 7); + assertEquals(mysql.getProperty("connectTimeout"), "7000"); + assertEquals(mysql.getProperty("socketTimeout"), "7000"); + } + + /** The connection string holds the credentials of the backend: a stall report must not carry them. */ + @Test + public void testLoggedConnectionStringCarriesNoCredentials() throws Exception { + assertEquals(CachedConnection.safeUrl("jdbc:postgresql://h:5432/db?user=u&password=secret"), "jdbc:postgresql://h:5432/db"); + assertEquals(CachedConnection.safeUrl("jdbc:sqlserver://h:1433;databaseName=db;password=secret"), "jdbc:sqlserver://h:1433"); + assertEquals(CachedConnection.safeUrl("jdbc:oracle:thin:scott/secret@//h:1521/svc"), "jdbc:oracle:@//h:1521/svc"); + assertEquals(CachedConnection.safeUrl("jdbc:mysql://u:secret@h:3306/db"), "jdbc:mysql:@h:3306/db"); + assertEquals(CachedConnection.safeUrl("jdbc:oracle:thin:@//h:1521/svc"), "jdbc:oracle:@//h:1521/svc"); + } + + private static SQLException tooManyConnections() { + // 53300, too_many_connections, of the insufficient_resources class + return new SQLException("sorry, too many clients already", "53300"); + } + + /** A connection string of every dialect pointing at one host and port. */ + private static String[] urlsOf(int port) { + return new String[]{ + "jdbc:postgresql://127.0.0.1:" + port + "/opendj?user=opendj&password=opendj", + "jdbc:mysql://127.0.0.1:" + port + "/opendj?user=opendj&password=opendj", + "jdbc:oracle:thin:opendj/opendj@//127.0.0.1:" + port + "/free", + "jdbc:sqlserver://127.0.0.1:" + port + ";databaseName=opendj;user=opendj;password=opendj;encrypt=false" + }; + } + + private static int closedPort() throws Exception { + try (final ServerSocket socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + return socket.getLocalPort(); + } // closed again: nothing listens there any more + } + + private static void assertElapsedWithinBound(long startedAt, long boundMs) { + final long elapsed = System.currentTimeMillis() - startedAt; + assertTrue(elapsed < boundMs + BOUND_MARGIN_MS, "gave up only after " + elapsed + " ms"); + } + + /** + * Stands in for a database whose answer to a connect is the point of the test: the vendor + * codes and SQL states below are what the retry has to tell apart, and no engine is needed to + * produce them. + */ + private static final class StubDriver implements Driver { + static final String PREFIX = "jdbc:opendj-stub:"; + static final int ALWAYS = -1; + + final AtomicInteger attempts = new AtomicInteger(); + private volatile SQLException failure; + private volatile int failuresLeft; + private volatile Connection answer; + + void failWith(SQLException failure, int times) { + this.failure = failure; + this.failuresLeft = times; + this.answer = null; + this.attempts.set(0); + } + + void answerWith(Connection answer) { + this.failure = null; + this.failuresLeft = 0; + this.answer = answer; + this.attempts.set(0); + } + + @Override + public Connection connect(String url, Properties info) throws SQLException { + if (!acceptsURL(url)) { + return null; + } + attempts.incrementAndGet(); + if (failuresLeft != 0) { + if (failuresLeft > 0) { + failuresLeft--; + } + throw failure; + } + if (answer != null) { + return answer; + } + final Connection con = mock(Connection.class); + when(con.isValid(anyInt())).thenReturn(true); + return con; + } + + @Override + public boolean acceptsURL(String url) { + return url != null && url.startsWith(PREFIX); + } + + @Override + public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) { + return new DriverPropertyInfo[0]; + } + + @Override + public int getMajorVersion() { + return 1; + } + + @Override + public int getMinorVersion() { + return 0; + } + + @Override + public boolean jdbcCompliant() { + return false; + } + + @Override + public Logger getParentLogger() { + return Logger.getLogger(StubDriver.class.getName()); + } + } +} diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java index b650301554..f925dc5171 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java @@ -140,6 +140,27 @@ public void cleanUp() throws Exception { protected abstract String getJdbcUrl(); + /** + * The second property bounding a login is a socket read timeout on mysql, oracle and sql + * server: in force for the whole life of the connection it would fail every statement slower + * than it - an import batch, the statistics of a freshly loaded table - so it has to be lifted + * as soon as the login is through (#872). + */ + @Test + public void testLoginBoundDoesNotOutliveTheLogin() throws Exception { + final String url = createBackendCfg().getDBDirectory(); + System.setProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY, "2"); + try { + // a pooled connection would be handed back without being established again + CachedConnection.cached.invalidate(url); + try (final Connection con = CachedConnection.getConnection(url)) { + assertEquals(con.getNetworkTimeout(), 0, "the read bound of the login is still in force"); + } + } finally { + System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY); + } + } + private static ByteString key(int i) { return ByteString.valueOfUtf8(String.format("key%02d", i)); } From ae2964a4d7bbce9071fb945431c7778b3029d9c0 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Wed, 19 Aug 2026 17:20:39 +0300 Subject: [PATCH 2/4] [#878] Bound the JDBC connection pool and expire its connections one by one The pool held its connections in an unbounded queue behind a cache entry keyed by the connection string, and four things followed from that shape. Nothing limited how many connections a backend opened. The queue was unbounded and the cache had no maximum size, so a burst of concurrent operations opened as many connections as there were threads asking, and the only ceiling left was the max_connections of the database - which the backend then treats as a condition to wait out, turning a burst into a burst of connect attempts against a server already at its limit. The TTL sat on the entry rather than on a connection. expireAfterAccess is reset by every read of the entry and both the borrow and the return read it, so under continuous traffic the entry never expired and neither did anything in it: the peak count of a burst stayed open for as long as the backend saw any traffic at all. Expiry was lazy, and the case it exists for is the one it missed. The cache was built without a scheduler, so an entry was only expired by a later cache operation, and a backend that has gone idle - the only situation in which it could expire at all - performs none. A closed backend released nothing: close() only flipped the storage status. The entry is replaced by a pool of its own. A semaphore bounds it at org.openidentityplatform.opendj.jdbc.pool.max, defaulting to the shape of the server's worker thread pool - max(16, 2 x CPU) - since an operation holds one connection for its duration; a borrow above the bound waits for a returned connection until the deadline of pool.timeout and then reports which bound it hit, rather than opening one more. Idle connections are kept most recently returned first, each carrying the time it was returned, so the TTL is a property of a connection: the ones a burst opened sink to the bottom, where the borrow and a sweeper thread running every half TTL find them - the sweep needs no borrow behind it. The return goes straight to the pool the connection came from instead of through a lookup of its connection string, which could hand back an entry that was evicted between the two, leaving the connection in a queue nothing referred to any more. A borrow made while the same thread already holds a connection passes the bound. PersistentCompressedSchema.store() opens a write of its own - the definition has to commit independently of the entry - and is reached from inside a transaction by EntryContainer.importEntry and EntryContainer.modifyDN, both of which encode the entry inside it; making the second borrow wait for the first would wait for this very thread. Such a connection holds no permit and is closed rather than pooled on return, so the pool does not grow past its bound. JDBCStorage registers as a user of the pool of its connection string when it opens and gives it up when it closes, the connections being released once the last user is gone. Reference counted because a pool belongs to a database rather than to a backend: two backends may address one database, and closing one of them must not take the connections of the other with it. ImporterImpl.close() now returns its connection from a finally block. A commit that failed skipped the close, so the connection was leaked along with the failure. --- .../backends/jdbc/CachedConnection.java | 362 ++++++++++++++++-- .../server/backends/jdbc/JDBCStorage.java | 26 +- .../jdbc/CachedConnectionTestCase.java | 179 ++++++++- .../opends/server/backends/jdbc/TestCase.java | 2 +- 4 files changed, 525 insertions(+), 44 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java index 874b8162fe..91f82bdf70 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java @@ -15,14 +15,10 @@ */ package org.opends.server.backends.jdbc; -import com.github.benmanes.caffeine.cache.Caffeine; -import com.github.benmanes.caffeine.cache.LoadingCache; -import com.github.benmanes.caffeine.cache.RemovalCause; import org.forgerock.i18n.LocalizableMessage; import org.forgerock.i18n.slf4j.LocalizedLogger; import java.sql.*; -import java.time.Duration; import java.util.Locale; import java.util.Map; import java.util.Properties; @@ -47,10 +43,35 @@ public class CachedConnection implements Connection { /** Bound of the validation of a pooled connection: isValid(0) means "no timeout" in the JDBC contract. */ static final int VALIDATION_TIMEOUT_SECONDS = 5; + /** The greatest number of connections one pool holds to one database; 0 for no bound. */ + static final String POOL_MAX_PROPERTY = "org.openidentityplatform.opendj.jdbc.pool.max"; + /** + * Sized like the worker thread pool of the server ({@code Platform.computeNumberOfThreads(16, 2)}), + * since an operation borrows one connection for its duration: the bound is there to keep a burst + * from opening as many connections as the database will accept, not to throttle steady traffic. + */ + static final int DEFAULT_POOL_MAX = Math.max(16, Runtime.getRuntime().availableProcessors() * 2); + + /** How long a borrow waits for a connection to be returned before looking at the pool again. */ + private static final long POOL_FULL_POLL_MS = 250; + /** The sweep runs at half the TTL, and no more often than this. */ + private static final long MIN_SWEEP_INTERVAL_MS = 1000; + static final long MAX_BACKOFF_MS = 1000; static final long STALL_WARNING_AFTER_MS = 1000; static final long STALL_WARNING_INTERVAL_MS = 10000; + /** + * How many connections the current thread holds. A borrow made while one is already held may + * exceed the bound of the pool, because the two are held at the same time and waiting for the + * first to be returned would wait forever: {@code PersistentCompressedSchema.store()} opens a + * write of its own - the definition has to commit independently of the entry - and reaches it + * from inside a transaction on two paths, {@code EntryContainer.importEntry} and + * {@code EntryContainer.modifyDN}, both of which encode the entry inside the transaction. + * Such a connection is never pooled on return, so the pool does not grow past its bound. + */ + private static final ThreadLocal held = ThreadLocal.withInitial(() -> new int[1]); + // setNetworkTimeout() takes the executor its timeout handling runs on: the drivers it is used // with here only set a socket option in it, so it costs a call rather than a thread. private static final Executor DIRECT_EXECUTOR = Runnable::run; @@ -60,29 +81,246 @@ public class CachedConnection implements Connection { final Connection parent; - static LoadingCache> cached = Caffeine.newBuilder() - .expireAfterAccess(Duration.ofMillis(getCacheTtlMillis())) - .removalListener((String key, BlockingQueue value, RemovalCause cause) -> { - for (CachedConnection con : value) { - try { - if (!con.isClosed()) { - con.parent.close(); - } - } catch (SQLException e) { - // ignore - } - } - }) - .build(conStr -> new LinkedBlockingQueue<>()); + /** The pool this connection belongs to, held directly so that the return needs no lookup. */ + private final Pool pool; + /** Whether this connection holds a permit of its pool: a reentrant borrow does not. */ + private final boolean metered; + /** The thread that borrowed it, so that a return on another thread does not corrupt the count. */ + private volatile Thread owner; + /** When it was last returned to the pool, which is what the TTL is measured from. */ + volatile long returnedAtMillis; + private final AtomicBoolean permitReleased = new AtomicBoolean(); + + /** The pool of every connection string in use, kept until the last storage using it closes. */ + static final ConcurrentMap pools = new ConcurrentHashMap<>(); + + /** The sweep that closes connections nothing has borrowed for the TTL, started with the first pool. */ + private static volatile ScheduledExecutorService sweeper; /** * Returns the time after which an idle pooled connection is closed, as configured by the * {@value #TTL_PROPERTY} system property. An invalid value is ignored in favor of the default. + *

+ * Read on every borrow and every sweep rather than once, so that it can be changed on a running + * server the way the bounds of a borrow can. */ - private static long getCacheTtlMillis() { + static long getCacheTtlMillis() { return getNonNegativeProperty(TTL_PROPERTY, DEFAULT_TTL_MS); } + /** The pool of a connection string, created on first use. */ + static Pool poolOf(String connectionString) { + final Pool pool = pools.computeIfAbsent(connectionString, Pool::new); + startSweeper(); + return pool; + } + + private static void startSweeper() { + if (sweeper != null) { + return; + } + synchronized (pools) { + if (sweeper == null) { + final ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(runnable -> { + final Thread thread = new Thread(runnable, "JDBC backend connection pool sweeper"); + thread.setDaemon(true); + return thread; + }); + final long interval = Math.max(MIN_SWEEP_INTERVAL_MS, getCacheTtlMillis() / 2); + service.scheduleWithFixedDelay(CachedConnection::sweep, interval, interval, TimeUnit.MILLISECONDS); + sweeper = service; + } + } + } + + // Expiry has to happen without a borrow behind it. Caffeine was left without a scheduler, so an + // entry was only ever expired by a later cache operation - and a backend that has gone idle, + // the one case the TTL exists for, performs none (issue #878). + private static void sweep() { + final long ttlMillis = getCacheTtlMillis(); + for (final Pool pool : pools.values()) { + try { + pool.sweep(ttlMillis); + } catch (RuntimeException e) { + logger.traceException(e); + } + } + } + + /** + * Registers a storage as a user of the pool of a connection string. Reference counted because a + * pool belongs to a database rather than to a backend: two backends may address one database, + * and closing one of them must not take the connections of the other with it. + */ + static void openPool(String connectionString) { + poolOf(connectionString).addUser(); + } + + /** Unregisters a storage; the connections are released once the last user is gone. */ + static void closePool(String connectionString) { + final Pool pool = pools.get(connectionString); + if (pool != null) { + pool.removeUser(); + } + } + + /** Closes every idle connection of a connection string, leaving the pool usable. */ + static void invalidate(String connectionString) { + final Pool pool = pools.get(connectionString); + if (pool != null) { + pool.drainIdle(); + } + } + + /** + * The connections of one connection string. + *

+ * This replaces the cache entry that used to hold them. That one carried the TTL on the pool + * rather than on a connection - {@code expireAfterAccess} keyed by the connection string, reset + * by every borrow and every return - so under continuous traffic nothing ever expired and the + * peak count of a burst stayed open for as long as the backend saw any traffic at all. It also + * had no bound, so the only ceiling on the connections of a backend was the {@code + * max_connections} of the database itself (issue #878). + */ + static final class Pool { + final String connectionString; + /** Idle connections, most recently returned first: the ones a burst opened sink to the bottom, where the sweep finds them. */ + private final LinkedBlockingDeque idle = new LinkedBlockingDeque<>(); + /** One permit per live connection, borrowed or idle. Sized once: this is how large the pool may grow, not a rate. */ + private final Semaphore permits; + private final int max; + /** Open storages using this pool, guarded by this. */ + private int users; + /** + * Set when the last storage using this pool closed. A pool no storage ever registered with - + * a borrow made straight through {@link CachedConnection#getConnection}, as the tests do - + * is not closed and pools normally; only one that had a user and lost it stops keeping + * connections for a borrower that is not going to come. + */ + private volatile boolean closed; + + Pool(String connectionString) { + this.connectionString = connectionString; + final long configured = getNonNegativeProperty(POOL_MAX_PROPERTY, DEFAULT_POOL_MAX); + this.max = (configured == 0 || configured > Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int) configured; + this.permits = new Semaphore(max); + } + + int max() { + return max; + } + + int idleCount() { + return idle.size(); + } + + /** The connections this pool holds, borrowed and idle together. */ + int liveCount() { + return max - permits.availablePermits(); + } + + synchronized void addUser() { + users++; + closed = false; + } + + void removeUser() { + final boolean wasLast; + synchronized (this) { + wasLast = users > 0 && --users == 0; + if (wasLast) { + closed = true; + } + } + if (wasLast) { + // Outside the monitor: closing a connection is a round trip, and an open of the + // same database has no reason to wait behind it. The borrowed ones are not here to + // be closed - give() closes them when they come back, since a pool nobody uses must + // not keep them for a borrower that is not going to come. + logger.trace(LocalizableMessage.raw("releasing %d pooled connections of %s: its last user closed", + idle.size(), safeUrl(connectionString))); + drainIdle(); + } + } + + void drainIdle() { + for (CachedConnection con = idle.pollFirst(); con != null; con = idle.pollFirst()) { + destroy(con); + } + } + + /** + * Takes a connection out of the pool, waiting up to waitMs for one to be returned, and + * discarding the ones that are broken or have been idle for longer than the TTL. + */ + CachedConnection pollIdle(long waitMs, long ttlMillis) throws InterruptedException { + long remainingWait = waitMs; + while (true) { + final long polledAt = System.currentTimeMillis(); + final CachedConnection con = idle.pollFirst(remainingWait, TimeUnit.MILLISECONDS); + if (con == null) { + return null; + } + if (System.currentTimeMillis() - con.returnedAtMillis <= ttlMillis && isUsable(con)) { + return con; + } + destroy(con); + remainingWait -= System.currentTimeMillis() - polledAt; + if (remainingWait <= 0) { + // one more look, since a connection may have been returned in the meantime + remainingWait = 0; + } + } + } + + /** Takes the right to hold one more connection, or reports that the pool is full. */ + boolean tryReserve() { + return permits.tryAcquire(); + } + + void cancelReservation() { + permits.release(); + } + + /** Hands a connection back, closing it rather than pooling it when it may not be kept. */ + void give(CachedConnection con) { + // An unmetered connection holds no permit, so pooling it would put the pool one over its + // bound for good; and a closed pool has nobody left to hand it to. + if (con.metered && !closed) { + addIdle(con); + if (closed) { + // The last user left while this one was on its way back, so it missed the drain. + drainIdle(); + } + } else { + destroy(con); + } + } + + /** Puts a connection into the pool. The caller must hold the right to keep it there. */ + void addIdle(CachedConnection con) { + con.returnedAtMillis = System.currentTimeMillis(); + idle.addFirst(con); + } + + void destroy(CachedConnection con) { + closeQuietly(con.parent); + con.releasePermit(); + } + + void sweep(long ttlMillis) { + final long deadline = System.currentTimeMillis() - ttlMillis; + // From the tail: the least recently returned connection is the first to have expired, + // and once one has not, neither has anything in front of it. + for (CachedConnection con = idle.peekLast(); con != null; con = idle.peekLast()) { + if (con.returnedAtMillis > deadline || !idle.removeLastOccurrence(con)) { + return; + } + destroy(con); + } + } + } + /** * Returns the value of a numeric system property, ignoring a value that is not a non-negative * number in favor of the default. @@ -211,9 +449,38 @@ private static boolean containsParameter(String connectionString, String propert } final String connectionString; + + /** A connection outside the accounting of its pool: it holds no permit and is never pooled. */ public CachedConnection(String connectionString, Connection parent) { + this(connectionString, parent, poolOf(connectionString), false); + } + + CachedConnection(String connectionString, Connection parent, Pool pool, boolean metered) { this.connectionString = connectionString; this.parent = parent; + this.pool = pool; + this.metered = metered; + } + + /** Gives back the right to hold this connection, once and only if it was taken. */ + void releasePermit() { + if (metered && permitReleased.compareAndSet(false, true)) { + pool.cancelReservation(); + } + } + + /** Records that the borrowing thread holds this connection, so a borrow nested in it is recognized. */ + private static CachedConnection borrowed(CachedConnection con) { + con.owner = Thread.currentThread(); + held.get()[0]++; + return con; + } + + private static void released() { + final int[] depth = held.get(); + if (depth[0] > 0) { + depth[0]--; + } } /** @@ -223,25 +490,47 @@ public CachedConnection(String connectionString, Connection parent) { * not answer into a hang rather than into an error the caller can report. */ static Connection getConnection(String connectionString) throws Exception { + final Pool pool = poolOf(connectionString); final ConnectDialect dialect = ConnectDialect.of(connectionString); final long connectTimeoutSeconds = Math.min( getNonNegativeProperty(CONNECT_TIMEOUT_PROPERTY, DEFAULT_CONNECT_TIMEOUT_SECONDS), Integer.MAX_VALUE / 1000); final long poolTimeoutSeconds = getNonNegativeProperty(POOL_TIMEOUT_PROPERTY, DEFAULT_POOL_TIMEOUT_SECONDS); + final long ttlMillis = getCacheTtlMillis(); final long startedAt = System.currentTimeMillis(); final long deadline = (poolTimeoutSeconds == 0 || poolTimeoutSeconds >= Long.MAX_VALUE / 1000) ? Long.MAX_VALUE : startedAt + poolTimeoutSeconds * 1000; + // A thread already holding a connection is not made to wait for one: the two are held at + // the same time, so waiting for the first to come back would wait for itself. + final boolean reentrant = held.get()[0] > 0; long waitMs = 0; long backoffMs = 0; int attempts = 0; while (true) { - final CachedConnection pooled = poll(connectionString, waitMs); + final CachedConnection pooled = pool.pollIdle(waitMs, ttlMillis); if (pooled != null) { - return pooled; + return borrowed(pooled); + } + if (!reentrant && !pool.tryReserve()) { + // The pool holds as many connections as it may: only a returned one can serve this + // borrow now, and the deadline decides how long that is worth waiting for. This is + // the point of the bound - without it the borrow would open one more connection, + // and the only ceiling left would be the max_connections of the database itself. + final long remaining = deadline - System.currentTimeMillis(); + if (remaining <= 0) { + throw new SQLTimeoutException("no connection to " + safeUrl(connectionString) + + " could be borrowed within " + poolTimeoutSeconds + "s: all " + pool.max() + + " connections of the pool are in use (raise " + POOL_MAX_PROPERTY + " to allow more)"); + } + waitMs = Math.min(POOL_FULL_POLL_MS, remaining); + continue; } attempts++; try { - return connect(connectionString, dialect, connectTimeoutSeconds); + return borrowed(connect(connectionString, dialect, connectTimeoutSeconds, pool, !reentrant)); } catch (SQLException e) { + if (!reentrant) { + pool.cancelReservation(); + } // A database that accepts no further connection is the one failure worth waiting // out: one of ours is going to come back to the pool. Everything else - a password // that is not accepted, a database that is down, a driver that is not on the @@ -262,19 +551,6 @@ static Connection getConnection(String connectionString) throws Exception { } } - /** Takes a usable connection out of the pool, waiting up to waitMs for one to be returned to it. */ - private static CachedConnection poll(String connectionString, long waitMs) throws InterruptedException { - CachedConnection con = cached.get(connectionString).poll(waitMs, TimeUnit.MILLISECONDS); - while (con != null) { - if (isUsable(con)) { - return con; - } - closeQuietly(con.parent); - con = cached.get(connectionString).poll(); - } - return null; - } - private static boolean isUsable(CachedConnection con) { try { // The validation needs a bound of its own: isValid(0) means "no timeout" in the JDBC @@ -286,8 +562,8 @@ private static boolean isUsable(CachedConnection con) { } } - private static CachedConnection connect(String connectionString, ConnectDialect dialect, long connectTimeoutSeconds) - throws SQLException { + private static CachedConnection connect(String connectionString, ConnectDialect dialect, long connectTimeoutSeconds, + Pool pool, boolean metered) throws SQLException { // A driver is free to write into the map it is handed, so it gets one of its own. final Properties properties = new Properties(); final boolean readBoundSet = dialect != null && connectTimeoutSeconds > 0 @@ -304,7 +580,7 @@ private static CachedConnection connect(String connectionString, ConnectDialect closeQuietly(conNew); throw e; } - return new CachedConnection(connectionString, conNew); + return new CachedConnection(connectionString, conNew, pool, metered); } // The second bound of the login is a socket read timeout on mysql, oracle and sql server, in @@ -430,15 +706,21 @@ public void rollback() throws SQLException { @Override public void close() throws SQLException { + if (owner == Thread.currentThread()) { + released(); + } try { rollback(); } catch (SQLException e) { // A connection that cannot be rolled back must not be handed to the next borrower - // and must not be dropped on the floor either: nothing else holds it any more. - closeQuietly(parent); + pool.destroy(this); throw e; } - cached.get(connectionString).add(this); + // Straight to the pool it came from rather than through a lookup of its connection string: + // the entry the lookup returned could be evicted between the two, leaving the connection in + // a queue nothing referred to any more - never handed out, never closed (issue #878). + pool.give(this); } @Override diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java index d1d8055314..34db96d0c3 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java @@ -41,6 +41,7 @@ import java.sql.*; import java.util.*; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; import static org.opends.server.backends.pluggable.spi.StorageUtils.addErrorMessage; import static org.opends.server.util.StaticUtils.stackTraceToSingleLineString; @@ -145,11 +146,26 @@ Connection getConnection() throws Exception { AccessMode accessMode=AccessMode.READ_ONLY; + + // Whether this storage counts as a user of the pool of its connection string. The pool belongs + // to the database rather than to this backend - two backends may address one database - so it + // is reference counted, and this flag keeps an open() or a close() that comes twice from + // counting twice (issue #878). + private final AtomicBoolean poolRegistered=new AtomicBoolean(); + @Override public void open(AccessMode accessMode) throws Exception { + if (poolRegistered.compareAndSet(false, true)) { + CachedConnection.openPool(config.getDBDirectory()); + } try (final Connection con=getConnection()) { this.accessMode = accessMode; storageStatus = StorageStatus.working(); + } catch (Exception e) { + if (poolRegistered.compareAndSet(true, false)) { + CachedConnection.closePool(config.getDBDirectory()); + } + throw e; } } @@ -166,6 +182,12 @@ public void close() { // that it is not reissued for every tree on every open; disabling and re-enabling the // backend is the way to try again once the privilege has been granted unstampableTrees.clear(); + // A closed backend has no use for its connections. They used to stay open - close() only + // flipped the status - so disabling or removing a JDBC backend left them behind, and with + // nothing left to expire the pool entry they could stay open for good (issue #878). + if (poolRegistered.compareAndSet(true, false)) { + CachedConnection.closePool(config.getDBDirectory()); + } } final LoadingCache tree2table = Caffeine.newBuilder() @@ -1532,7 +1554,9 @@ public void close() { }else { updateTableStatistics(con, writtenTrees); } - } finally { // the pooled connection must be returned even when the commit or a statistics statement throws + } finally { + // Back to the pool even when the commit or a statistics statement failed: nothing + // else holds this connection, so leaving it behind would leak it along with the failure. try { con.close(); } finally { diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java index 1590536363..bc96b5a716 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java @@ -46,6 +46,7 @@ import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertNotSame; import static org.testng.Assert.assertNull; import static org.testng.Assert.assertSame; import static org.testng.Assert.assertTrue; @@ -81,6 +82,180 @@ public void deregisterStubDriver() throws Exception { public void clearProperties() { System.clearProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY); System.clearProperty(CachedConnection.POOL_TIMEOUT_PROPERTY); + System.clearProperty(CachedConnection.POOL_MAX_PROPERTY); + System.clearProperty(CachedConnection.TTL_PROPERTY); + } + + /** + * Nothing used to limit how many connections a backend opened: the pool was an unbounded queue + * behind a cache with no maximum size, so a burst of concurrent operations opened as many + * connections as there were threads asking, and the only ceiling left was the max_connections of + * the database itself (#878). + */ + @Test(timeOut = 120000) + public void testThePoolDoesNotGrowPastItsBound() throws Exception { + final String url = StubDriver.PREFIX + "bounded"; + System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "2"); + System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "1"); + stub.answerWith(null); + + // One thread per borrow, as the worker threads of the server are: two borrows on one thread + // are nested by definition, and a nested one is allowed past the bound on purpose. + final Connection first = borrowOnAThreadOfItsOwn(url); + final Connection second = borrowOnAThreadOfItsOwn(url); + assertEquals(CachedConnection.poolOf(url).liveCount(), 2); + try { + borrowOnAThreadOfItsOwn(url); + fail("a third connection was opened past the bound of two"); + } catch (ExecutionException e) { + assertTrue(e.getCause() instanceof SQLTimeoutException, String.valueOf(e.getCause())); + assertTrue(e.getCause().getMessage().contains("all 2 connections"), e.getCause().getMessage()); + } + + // The bound waits for a returned connection rather than refusing outright: it is a ceiling + // on the connections held, not on the operations served. + first.close(); + final Connection third = borrowOnAThreadOfItsOwn(url); + assertSame(third, first); + third.close(); + second.close(); + CachedConnection.invalidate(url); + } + + /** Borrows the way the server does, one operation to a thread. */ + private static Connection borrowOnAThreadOfItsOwn(String url) throws Exception { + final FutureTask borrow = new FutureTask<>(() -> CachedConnection.getConnection(url)); + final Thread thread = new Thread(borrow, "borrow-" + url); + thread.setDaemon(true); + thread.start(); + return borrow.get(120, TimeUnit.SECONDS); + } + + /** + * A borrow made while this thread already holds a connection must not wait for the bound: the + * two are held at once, so it would wait for itself. PersistentCompressedSchema.store() opens a + * write of its own and is reached from inside a transaction by EntryContainer.importEntry and + * EntryContainer.modifyDN, both of which encode the entry inside it. + */ + @Test(timeOut = 120000) + public void testABorrowNestedInAnotherMayPassTheBound() throws Exception { + final String url = StubDriver.PREFIX + "reentrant"; + System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "1"); + System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "1"); + stub.answerWith(null); + + final Connection outer = CachedConnection.getConnection(url); + final Connection nested = CachedConnection.getConnection(url); + assertNotSame(nested, outer); + + // It holds no permit of the pool, so pooling it would leave the pool one connection over + // its bound for good: it is closed instead. + nested.close(); + verify(((CachedConnection) nested).parent).close(); + assertEquals(CachedConnection.poolOf(url).idleCount(), 0); + + outer.close(); + assertEquals(CachedConnection.poolOf(url).idleCount(), 1); + CachedConnection.invalidate(url); + } + + /** + * The TTL used to sit on the pool rather than on a connection - keyed by the connection string, + * and touched by every borrow and every return - so under continuous traffic nothing in it ever + * expired (#878). + */ + @Test(timeOut = 120000) + public void testAnIdleConnectionIsClosedAfterItsTtl() throws Exception { + final String url = StubDriver.PREFIX + "ttl"; + stub.answerWith(null); + + final CachedConnection first = (CachedConnection) CachedConnection.getConnection(url); + first.close(); + assertEquals(CachedConnection.poolOf(url).idleCount(), 1); + first.returnedAtMillis = System.currentTimeMillis() - 60000; + System.setProperty(CachedConnection.TTL_PROPERTY, "1000"); + + final Connection second = CachedConnection.getConnection(url); + + assertNotSame(second, first, "a connection idle far longer than the TTL was handed out"); + verify(first.parent).close(); + second.close(); + CachedConnection.invalidate(url); + } + + /** + * Expiry has to happen without a borrow behind it: the cache was built without a scheduler, so + * an entry was only ever expired by a later cache operation - and a backend that has gone idle, + * the one case the TTL exists for, performs none (#878). This is what the sweeper thread runs. + */ + @Test(timeOut = 120000) + public void testTheSweepClosesAnIdleConnectionWithNoBorrowBehindIt() throws Exception { + final String url = StubDriver.PREFIX + "sweep"; + stub.answerWith(null); + final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url); + con.close(); + con.returnedAtMillis = System.currentTimeMillis() - 60000; + final CachedConnection.Pool pool = CachedConnection.poolOf(url); + assertEquals(pool.idleCount(), 1); + + pool.sweep(1000); + + assertEquals(pool.idleCount(), 0); + verify(con.parent).close(); + assertEquals(pool.liveCount(), 0, "a swept connection kept its place in the pool"); + } + + /** A closed backend has no use for its connections; they used to be left open (#878). */ + @Test(timeOut = 120000) + public void testClosingTheLastUserReleasesTheConnections() throws Exception { + final String url = StubDriver.PREFIX + "release"; + stub.answerWith(null); + CachedConnection.openPool(url); + final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url); + con.close(); + assertEquals(CachedConnection.poolOf(url).idleCount(), 1); + + CachedConnection.closePool(url); + + assertEquals(CachedConnection.poolOf(url).idleCount(), 0); + verify(con.parent).close(); + assertEquals(CachedConnection.poolOf(url).liveCount(), 0); + } + + /** + * A pool belongs to a database rather than to a backend: two backends may address one database, + * and closing one of them must not take the connections of the other with it. + */ + @Test(timeOut = 120000) + public void testConnectionsSurviveWhileAnotherBackendStillUsesTheDatabase() throws Exception { + final String url = StubDriver.PREFIX + "shared"; + stub.answerWith(null); + CachedConnection.openPool(url); + CachedConnection.openPool(url); + final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url); + con.close(); + + CachedConnection.closePool(url); + assertEquals(CachedConnection.poolOf(url).idleCount(), 1, "the second backend lost its connections"); + + CachedConnection.closePool(url); + assertEquals(CachedConnection.poolOf(url).idleCount(), 0); + verify(con.parent).close(); + } + + /** A connection out on loan when the last backend closed is closed when it comes back. */ + @Test(timeOut = 120000) + public void testAConnectionReturnedAfterTheLastUserLeftIsClosed() throws Exception { + final String url = StubDriver.PREFIX + "return-after-close"; + stub.answerWith(null); + CachedConnection.openPool(url); + final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url); + + CachedConnection.closePool(url); + con.close(); + + verify(con.parent).close(); + assertEquals(CachedConnection.poolOf(url).idleCount(), 0); } /** @@ -228,7 +403,7 @@ public void testBrokenPooledConnectionIsDiscarded() throws Exception { final String url = StubDriver.PREFIX + "broken-pooled"; final Connection stale = mock(Connection.class); when(stale.isValid(anyInt())).thenReturn(false); - CachedConnection.cached.get(url).add(new CachedConnection(url, stale)); + CachedConnection.poolOf(url).addIdle(new CachedConnection(url, stale)); final Connection fresh = mock(Connection.class); when(fresh.isValid(anyInt())).thenReturn(true); stub.answerWith(fresh); @@ -256,7 +431,7 @@ public void testConnectionThatCannotBeRolledBackIsClosed() throws Exception { assertEquals(expected.getMessage(), "connection is closed"); } verify(parent).close(); - assertTrue(CachedConnection.cached.get(url).isEmpty(), "a connection that cannot be rolled back was pooled"); + assertEquals(CachedConnection.poolOf(url).idleCount(), 0, "a connection that cannot be rolled back was pooled"); } @Test diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java index f925dc5171..a6d8071e0d 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java @@ -152,7 +152,7 @@ public void testLoginBoundDoesNotOutliveTheLogin() throws Exception { System.setProperty(CachedConnection.CONNECT_TIMEOUT_PROPERTY, "2"); try { // a pooled connection would be handed back without being established again - CachedConnection.cached.invalidate(url); + CachedConnection.invalidate(url); try (final Connection con = CachedConnection.getConnection(url)) { assertEquals(con.getNetworkTimeout(), 0, "the read bound of the login is still in force"); } From 1a3e0344537ce1f4879b7afec04056e95b12f1d1 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Thu, 20 Aug 2026 16:19:11 +0300 Subject: [PATCH 3/4] [#878] Give back what a failed borrow took, and the rest of what the review found The pool kept no finally anywhere, and every release of a permit hung on the happy path or on catch (SQLException). DriverManager catches SQLException alone, so an unchecked failure of a driver - Connector/J hands a url with a "%" in it to URLDecoder, and this backend keeps its credentials in the url - left the reservation of the attempt behind. Nothing gives such a permit back: only a live connection carries one, and pools are static and never pruned, so after as many failures as the bound every borrow waited out pool.timeout and then reported that all connections were in use while the pool held none. The attempt now gives back what it took on every way out of it, connect() closes the connection it established whatever is thrown, and destroy() releases the permit from a finally. The reentrancy counter moved from one static ThreadLocal into the pool it belongs to. The deadlock its exemption guards against only exists within one pool, so a shared count judged a thread holding a connection to one database reentrant while it borrowed from another - passing the bound of a pool it held nothing of, and destroying the connection instead of pooling it, on every operation. The sweep hands its closes to an executor instead of running them. Every pool is swept on one thread and scheduleWithFixedDelay never overlaps its runs, so one close that did not return stopped the expiry of every pool in the JVM, silently: oracle logs off over the network, and the read bound of the login is lifted by then. close() became a no-op the second time, as the JDBC contract says, rather than putting the same connection into the pool twice. JDBCStorage keeps the connection string it registered with, since applyConfigurationChange replaces config and close() would otherwise release the pool of another database and leave its own with a user it never loses; the catch of open() now unregisters only what that call registered. The ImporterImpl constructor gives back the connection it borrowed and closes the storage it opened when a transaction below it throws - WriteableTransactionTransactionImpl rejects a storage opened READ_ONLY - since close() belongs to an object that was built. Its close() no longer lets the failure of the return replace the failure of the commit, which is the case where the commit failed on a broken connection. Five cases added, none of which needs a database. --- .../backends/jdbc/CachedConnection.java | 139 +++++++++++++----- .../server/backends/jdbc/JDBCStorage.java | 85 ++++++++--- .../jdbc/CachedConnectionTestCase.java | 125 +++++++++++++++- 3 files changed, 293 insertions(+), 56 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java index 91f82bdf70..e08eca67f9 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java @@ -61,17 +61,6 @@ public class CachedConnection implements Connection { static final long STALL_WARNING_AFTER_MS = 1000; static final long STALL_WARNING_INTERVAL_MS = 10000; - /** - * How many connections the current thread holds. A borrow made while one is already held may - * exceed the bound of the pool, because the two are held at the same time and waiting for the - * first to be returned would wait forever: {@code PersistentCompressedSchema.store()} opens a - * write of its own - the definition has to commit independently of the entry - and reaches it - * from inside a transaction on two paths, {@code EntryContainer.importEntry} and - * {@code EntryContainer.modifyDN}, both of which encode the entry inside the transaction. - * Such a connection is never pooled on return, so the pool does not grow past its bound. - */ - private static final ThreadLocal held = ThreadLocal.withInitial(() -> new int[1]); - // setNetworkTimeout() takes the executor its timeout handling runs on: the drivers it is used // with here only set a socket option in it, so it costs a call rather than a thread. private static final Executor DIRECT_EXECUTOR = Runnable::run; @@ -90,6 +79,8 @@ public class CachedConnection implements Connection { /** When it was last returned to the pool, which is what the TTL is measured from. */ volatile long returnedAtMillis; private final AtomicBoolean permitReleased = new AtomicBoolean(); + /** Whether it has been handed back already: JDBC makes close() on a closed connection a no-op. */ + private final AtomicBoolean returned = new AtomicBoolean(); /** The pool of every connection string in use, kept until the last storage using it closes. */ static final ConcurrentMap pools = new ConcurrentHashMap<>(); @@ -97,6 +88,9 @@ public class CachedConnection implements Connection { /** The sweep that closes connections nothing has borrowed for the TTL, started with the first pool. */ private static volatile ScheduledExecutorService sweeper; + /** Where the sweep closes what it reaped, so that a close which does not return keeps it: see {@link Pool#sweep}. */ + private static volatile Executor closer = DIRECT_EXECUTOR; + /** * Returns the time after which an idle pooled connection is closed, as configured by the * {@value #TTL_PROPERTY} system property. An invalid value is ignored in favor of the default. @@ -121,6 +115,14 @@ private static void startSweeper() { } synchronized (pools) { if (sweeper == null) { + // A thread per close in flight, and none while nothing is being closed. One thread + // shared by all of them would only move the head of the line, which is the point + // of not closing on the sweeper in the first place. + closer = Executors.newCachedThreadPool(runnable -> { + final Thread thread = new Thread(runnable, "JDBC backend connection pool closer"); + thread.setDaemon(true); + return thread; + }); final ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(runnable -> { final Thread thread = new Thread(runnable, "JDBC backend connection pool sweeper"); thread.setDaemon(true); @@ -138,9 +140,10 @@ private static void startSweeper() { // the one case the TTL exists for, performs none (issue #878). private static void sweep() { final long ttlMillis = getCacheTtlMillis(); + final Executor closeOn = closer; for (final Pool pool : pools.values()) { try { - pool.sweep(ttlMillis); + pool.sweep(ttlMillis, closeOn); } catch (RuntimeException e) { logger.traceException(e); } @@ -189,6 +192,21 @@ static final class Pool { /** One permit per live connection, borrowed or idle. Sized once: this is how large the pool may grow, not a rate. */ private final Semaphore permits; private final int max; + /** + * How many connections of this pool the current thread holds. A borrow made while one is + * already held may exceed the bound, because the two are held at the same time and waiting + * for the first to be returned would wait for this very thread: + * {@code PersistentCompressedSchema.store()} opens a write of its own - the definition has + * to commit independently of the entry - and {@code EntryContainer.modifyDN} reaches it + * from inside a transaction, having encoded the entry there. Such a connection is never + * pooled on return, so the pool does not grow past its bound. + *

+ * Counted per pool rather than per thread, because that deadlock only exists within one + * pool: a count shared by all of them would judge a thread holding a connection to one + * database reentrant while it borrows from another, passing the bound of a pool it holds + * nothing of and destroying the connection instead of pooling it, on every operation. + */ + private final ThreadLocal held = ThreadLocal.withInitial(() -> new int[1]); /** Open storages using this pool, guarded by this. */ private int users; /** @@ -210,6 +228,22 @@ int max() { return max; } + /** Whether the calling thread already holds a connection of this pool. */ + boolean heldByCurrentThread() { + return held.get()[0] > 0; + } + + void enter() { + held.get()[0]++; + } + + void leave() { + final int[] depth = held.get(); + if (depth[0] > 0) { + depth[0]--; + } + } + int idleCount() { return idle.size(); } @@ -304,11 +338,29 @@ void addIdle(CachedConnection con) { } void destroy(CachedConnection con) { - closeQuietly(con.parent); - con.releasePermit(); + try { + closeQuietly(con.parent); + } finally { + // However the close went, the pool holds one connection fewer. A permit not given + // back here is given back by nothing at all: only a live connection carries one, + // and this one is gone (issue #878). + con.releasePermit(); + } } void sweep(long ttlMillis) { + sweep(ttlMillis, DIRECT_EXECUTOR); + } + + /** + * Closes the connections nothing has borrowed for the TTL, handing each to the executor + * given rather than closing it here. The sweep of every pool shares one thread and + * {@code scheduleWithFixedDelay} never overlaps its runs, so one close that does not + * return would stop the expiry of every pool in the JVM - and silently, since only a + * thrown exception is logged. Oracle logs off over the network, and the read bound of the + * login has been lifted by then (issue #878). + */ + void sweep(long ttlMillis, Executor closeOn) { final long deadline = System.currentTimeMillis() - ttlMillis; // From the tail: the least recently returned connection is the first to have expired, // and once one has not, neither has anything in front of it. @@ -316,7 +368,12 @@ void sweep(long ttlMillis) { if (con.returnedAtMillis > deadline || !idle.removeLastOccurrence(con)) { return; } - destroy(con); + final CachedConnection expired = con; + try { + closeOn.execute(() -> destroy(expired)); + } catch (RuntimeException e) { // no thread to close it on: here rather than nowhere + destroy(expired); + } } } } @@ -472,17 +529,11 @@ void releasePermit() { /** Records that the borrowing thread holds this connection, so a borrow nested in it is recognized. */ private static CachedConnection borrowed(CachedConnection con) { con.owner = Thread.currentThread(); - held.get()[0]++; + con.returned.set(false); + con.pool.enter(); return con; } - private static void released() { - final int[] depth = held.get(); - if (depth[0] > 0) { - depth[0]--; - } - } - /** * Borrows a connection: a usable one out of the pool, or a newly established one. Bounded in * both phases - every operation of this backend, the open of a backend and the import @@ -501,7 +552,7 @@ static Connection getConnection(String connectionString) throws Exception { ? Long.MAX_VALUE : startedAt + poolTimeoutSeconds * 1000; // A thread already holding a connection is not made to wait for one: the two are held at // the same time, so waiting for the first to come back would wait for itself. - final boolean reentrant = held.get()[0] > 0; + final boolean reentrant = pool.heldByCurrentThread(); long waitMs = 0; long backoffMs = 0; int attempts = 0; @@ -525,12 +576,14 @@ static Connection getConnection(String connectionString) throws Exception { continue; } attempts++; + CachedConnection established = null; + boolean handedOff = false; try { - return borrowed(connect(connectionString, dialect, connectTimeoutSeconds, pool, !reentrant)); + established = connect(connectionString, dialect, connectTimeoutSeconds, pool, !reentrant); + final CachedConnection con = borrowed(established); + handedOff = true; + return con; } catch (SQLException e) { - if (!reentrant) { - pool.cancelReservation(); - } // A database that accepts no further connection is the one failure worth waiting // out: one of ours is going to come back to the pool. Everything else - a password // that is not accepted, a database that is down, a driver that is not on the @@ -547,6 +600,20 @@ static Connection getConnection(String connectionString) throws Exception { backoffMs = Math.min(backoffMs == 0 ? 1 : backoffMs * 2, MAX_BACKOFF_MS); waitMs = Math.min(backoffMs, remaining); warnStall(connectionString, attempts, startedAt, e); + } finally { + // What the attempt took is given back on every way out of it, not only on the + // SQLException a driver is supposed to throw. DriverManager catches SQLException + // alone, so an unchecked failure of a driver reaches here - Connector/J hands a url + // with a "%" in it to URLDecoder, and this backend keeps its credentials in the url + // - and a permit left behind is left behind for good: only a live connection + // carries one, and a failed attempt has none to give (issue #878). + if (!handedOff) { + if (established != null) { + pool.destroy(established); // the permit went with it, and comes back with it + } else if (!reentrant) { + pool.cancelReservation(); + } + } } } } @@ -576,9 +643,9 @@ private static CachedConnection connect(String connectionString, ConnectDialect if (readBoundSet) { relaxReadBound(conNew); } - } catch (SQLException e) { // nothing holds this connection yet: it would leak + } catch (Throwable t) { // nothing holds this connection yet: it would leak, whatever it is closeQuietly(conNew); - throw e; + throw t; } return new CachedConnection(connectionString, conNew, pool, metered); } @@ -659,8 +726,8 @@ static String safeUrl(String connectionString) { private static void closeQuietly(Connection con) { try { con.close(); - } catch (SQLException e) { - // ignore: it is on its way out anyway + } catch (SQLException | RuntimeException e) { + // ignore: it is on its way out anyway, and the caller has a permit to give back } } @@ -706,9 +773,15 @@ public void rollback() throws SQLException { @Override public void close() throws SQLException { + // JDBC makes close() on a closed connection a no-op, and this one has to be one: a second + // return would put the same connection into the pool twice, to be handed to two borrowers. + if (!returned.compareAndSet(false, true)) { + return; + } if (owner == Thread.currentThread()) { - released(); + pool.leave(); } + owner = null; try { rollback(); } catch (SQLException e) { diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java index 34db96d0c3..58eebdb5e8 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java @@ -153,22 +153,42 @@ Connection getConnection() throws Exception { // counting twice (issue #878). private final AtomicBoolean poolRegistered=new AtomicBoolean(); + // The connection string open() registered with. applyConfigurationChange() replaces config, so + // reading db-directory again at close() could give back the pool of a database this storage + // never registered with - leaving the one it did with a user it never loses (issue #878). + private volatile String poolConnectionString; + @Override public void open(AccessMode accessMode) throws Exception { - if (poolRegistered.compareAndSet(false, true)) { - CachedConnection.openPool(config.getDBDirectory()); + final boolean registeredHere=poolRegistered.compareAndSet(false, true); + if (registeredHere) { + poolConnectionString=config.getDBDirectory(); + CachedConnection.openPool(poolConnectionString); } try (final Connection con=getConnection()) { this.accessMode = accessMode; storageStatus = StorageStatus.working(); } catch (Exception e) { - if (poolRegistered.compareAndSet(true, false)) { - CachedConnection.closePool(config.getDBDirectory()); + // Only what this call registered is given back: an open that found the registration + // already made took nothing, and giving it back would release a pool still in use. + if (registeredHere) { + releasePool(); } throw e; } } + /** Gives up the registration of this storage with the pool of the database it opened. */ + private void releasePool() { + if (poolRegistered.compareAndSet(true, false)) { + final String registered=poolConnectionString; + poolConnectionString=null; + if (registered!=null) { + CachedConnection.closePool(registered); + } + } + } + private StorageStatus storageStatus = StorageStatus.lockedDown(LocalizableMessage.raw("closed")); @Override public StorageStatus getStorageStatus() { @@ -185,9 +205,7 @@ public void close() { // A closed backend has no use for its connections. They used to stay open - close() only // flipped the status - so disabling or removing a JDBC backend left them behind, and with // nothing left to expire the pool entry they could stay open for good (issue #878). - if (poolRegistered.compareAndSet(true, false)) { - CachedConnection.closePool(config.getDBDirectory()); - } + releasePool(); } final LoadingCache tree2table = Caffeine.newBuilder() @@ -1530,13 +1548,28 @@ public ImporterImpl() { throw new StorageRuntimeException(e); } } + // Nothing holds what this constructor takes until it returns: close() belongs to an + // object that was built, so a throw below - WriteableTransactionTransactionImpl rejects + // a storage opened READ_ONLY - would leave the connection borrowed and the storage this + // constructor opened open, with nobody left to give either back. + Connection borrowed=null; try { - con = getConnection(); + borrowed=getConnection(); + txr =new ReadableTransactionImpl(borrowed); + txw =new WriteableTransactionTransactionImpl(borrowed); + con = borrowed; + borrowed=null; }catch (Exception e){ - throw new StorageRuntimeException(e); + if (borrowed!=null) { + try { + borrowed.close(); + }catch (SQLException e2) {} + } + if (!isOpen) { + JDBCStorage.this.close(); + } + throw e instanceof StorageRuntimeException ? (StorageRuntimeException) e : new StorageRuntimeException(e); } - txr =new ReadableTransactionImpl(con); - txw =new WriteableTransactionTransactionImpl(con); } @Override @@ -1547,6 +1580,7 @@ public void aborted() { @Override public void close() { try { + SQLException failure=null; try { con.commit(); if (aborted) { @@ -1554,17 +1588,28 @@ public void close() { }else { updateTableStatistics(con, writtenTrees); } - } finally { - // Back to the pool even when the commit or a statistics statement failed: nothing - // else holds this connection, so leaving it behind would leak it along with the failure. - try { - con.close(); - } finally { - txw.stampSession.close(); + } catch (SQLException e) { + failure=e; + } + // Back to the pool even when the commit failed: nothing else holds this connection, + // so leaving it behind would leak it along with the failure. + try { + con.close(); + } catch (SQLException e) { + // The return rolls back, and the rollback fails on exactly the connection whose + // commit just did. The commit is what went wrong, so it stays the exception the + // caller sees and this one rides along with it instead of replacing it. + if (failure==null) { + failure=e; + }else { + failure.addSuppressed(e); } + } finally { + txw.stampSession.close(); + } + if (failure!=null) { + throw new StorageRuntimeException(failure); } - } catch (SQLException e) { - throw new StorageRuntimeException(e); } finally { if (!isOpen) { JDBCStorage.this.close(); diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java index bc96b5a716..49e2ff90df 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java @@ -29,6 +29,8 @@ import java.sql.DriverPropertyInfo; import java.sql.SQLException; import java.sql.SQLTimeoutException; +import java.util.ArrayList; +import java.util.List; import java.util.Properties; import java.util.concurrent.ExecutionException; import java.util.concurrent.FutureTask; @@ -258,6 +260,119 @@ public void testAConnectionReturnedAfterTheLastUserLeftIsClosed() throws Excepti assertEquals(CachedConnection.poolOf(url).idleCount(), 0); } + /** A backend closed and opened again pools its connections as before: addUser() clears the flag. */ + @Test(timeOut = 120000) + public void testABackendClosedAndOpenedAgainPoolsItsConnections() throws Exception { + final String url = StubDriver.PREFIX + "reopen"; + stub.answerWith(null); + CachedConnection.openPool(url); + CachedConnection.getConnection(url).close(); + CachedConnection.closePool(url); + assertEquals(CachedConnection.poolOf(url).idleCount(), 0); + + CachedConnection.openPool(url); + CachedConnection.getConnection(url).close(); + + assertEquals(CachedConnection.poolOf(url).idleCount(), 1, "a reopened backend stopped pooling its connections"); + CachedConnection.closePool(url); + } + + /** + * A connect that fails with something other than a SQLException must not cost the pool a + * permit. DriverManager catches SQLException alone, so an unchecked failure of a driver reaches + * the borrow: Connector/J hands a url with a "%" in it to URLDecoder, and this backend keeps + * its credentials in the url. Only a live connection carries a permit, so one left behind is + * left behind for good - after as many failures as the bound the pool would report that every + * connection is in use while holding none (#878). + */ + @Test(timeOut = 120000) + public void testAConnectFailingUncheckedCostsThePoolNothing() throws Exception { + final String url = StubDriver.PREFIX + "unchecked"; + System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "2"); + System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "1"); + final CachedConnection.Pool pool = CachedConnection.poolOf(url); + stub.failWith(new IllegalArgumentException("URLDecoder: Illegal hex characters in escape (%) pattern"), + StubDriver.ALWAYS); + + for (int i = 1; i <= 2 * pool.max(); i++) { + try { + CachedConnection.getConnection(url); + fail("the connect did not fail"); + } catch (IllegalArgumentException expected) { + // reported to the caller, as a configuration error has to be + } + assertEquals(pool.liveCount(), 0, "attempt " + i + " kept a permit of the pool"); + } + + // and the pool still serves, rather than reporting connections it does not hold as in use + stub.answerWith(null); + final Connection con = CachedConnection.getConnection(url); + assertNotNull(con); + con.close(); + CachedConnection.invalidate(url); + } + + /** + * The exemption of a nested borrow belongs to one pool: a thread holding a connection to one + * database holds nothing of another, so the bound of that other pool applies and its connection + * comes back to it rather than being closed. + */ + @Test(timeOut = 120000) + public void testHoldingAConnectionToOneDatabaseDoesNotExemptABorrowFromAnother() throws Exception { + final String first = StubDriver.PREFIX + "held-first"; + final String second = StubDriver.PREFIX + "held-second"; + stub.answerWith(null); + + final Connection held = CachedConnection.getConnection(first); + final Connection other = CachedConnection.getConnection(second); + assertEquals(CachedConnection.poolOf(second).liveCount(), 1, "the borrow passed the bound of the other pool"); + other.close(); + + assertEquals(CachedConnection.poolOf(second).idleCount(), 1, "the borrow was taken for a nested one and closed"); + held.close(); + CachedConnection.invalidate(first); + CachedConnection.invalidate(second); + } + + /** JDBC makes close() on a closed connection a no-op; a second return would pool the same one twice. */ + @Test(timeOut = 120000) + public void testASecondCloseDoesNotPoolTheConnectionTwice() throws Exception { + final String url = StubDriver.PREFIX + "double-close"; + stub.answerWith(null); + final Connection con = CachedConnection.getConnection(url); + con.close(); + con.close(); + + assertEquals(CachedConnection.poolOf(url).idleCount(), 1, "one connection was pooled twice"); + CachedConnection.invalidate(url); + } + + /** + * What the sweeper runs hands the close elsewhere instead of running it. The sweep of every + * pool shares one thread and scheduleWithFixedDelay never overlaps its runs, so one close that + * does not return would stop the expiry of every pool in the JVM, silently (#878). + */ + @Test(timeOut = 120000) + public void testTheSweepDoesNotCloseOnTheSweeperThread() throws Exception { + final String url = StubDriver.PREFIX + "sweep-elsewhere"; + stub.answerWith(null); + final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url); + con.close(); + con.returnedAtMillis = System.currentTimeMillis() - 60000; + final CachedConnection.Pool pool = CachedConnection.poolOf(url); + final List handedOff = new ArrayList<>(); + + pool.sweep(1000, handedOff::add); + + assertEquals(pool.idleCount(), 0, "the expired connection kept its place in the pool"); + verify(con.parent, never()).close(); + assertEquals(handedOff.size(), 1); + + handedOff.get(0).run(); + verify(con.parent).close(); + assertEquals(pool.liveCount(), 0, "a swept connection kept its permit"); + } + /** * A driver that is not on the classpath - the JDBC backend needs one dropped into * lib/extensions by hand - is a configuration error the caller has to see. Retried, it is @@ -547,11 +662,12 @@ private static final class StubDriver implements Driver { static final int ALWAYS = -1; final AtomicInteger attempts = new AtomicInteger(); - private volatile SQLException failure; + /** A SQLException, or the unchecked failure a driver is free to throw at DriverManager instead. */ + private volatile Throwable failure; private volatile int failuresLeft; private volatile Connection answer; - void failWith(SQLException failure, int times) { + void failWith(Throwable failure, int times) { this.failure = failure; this.failuresLeft = times; this.answer = null; @@ -575,7 +691,10 @@ public Connection connect(String url, Properties info) throws SQLException { if (failuresLeft > 0) { failuresLeft--; } - throw failure; + if (failure instanceof SQLException) { + throw (SQLException) failure; + } + throw (RuntimeException) failure; } if (answer != null) { return answer; From 642729ffc33235d356a68da4f8b747456ca04e06 Mon Sep 17 00:00:00 2001 From: Valera V Harseko Date: Fri, 21 Aug 2026 11:10:04 +0300 Subject: [PATCH 4/4] [#878] Keep a borrow inside its deadline, and give its connection back whatever fails ImporterImpl.close() guarded its commit against SQLException alone, and the return of the connection sat below that catch rather than in a finally. An Error out of a bulk import, or a driver failing unchecked, skipped both the return and the close of the stamp session - and only that return gives back the permit the borrow took. A pool is never removed from the static map, so the permit was gone for the life of the server, and enough imports walk the bound of the pool down to nothing, after which every borrow waits out pool.timeout and then reports that all connections are in use. The commit is guarded against everything now, and the failure of the return rides along with what escaped instead of replacing it. A borrow uses the connection string open() registered with, rather than the one config carries at the time. db-directory reaches the listener of a running backend - applyConfigurationChange takes it, isConfigurationChangeAcceptable refuses nothing, and the component-restart admin action renders a message rather than holding the change back - so a borrow after such a change drew from a pool no storage had registered with, while the pool this one did register with kept a user that never borrows: the leak of #878 back through the configuration. And that unregistered pool is drained the moment another backend which did register with it closes, leaving this one to reconnect for every operation. pollIdle() stops at the deadline of the borrow. pollFirst(0, MS) is a poll of no duration but still hands out whatever the deque holds, so the clamp to zero only made the rest of the polls non-blocking and the loop drained the deque whatever the deadline said. A connection whose socket is half-open - a moved VIP, a firewall that dropped the idle sockets - costs the validation timeout to discard, and the pool holds as many of those as its bound allows: 80s of validation against a 60s pool.timeout by default, 320s on 32 cores, before the connect that follows had even started. The validation is bounded by what is left of the borrow as well. A sweep gives up on the one connection a borrow took from under it rather than on the whole cycle, which used to leave everything else that had expired open until the next one. An Error out of a sweep no longer reaches scheduleWithFixedDelay, which never runs a task that threw again: the expiry of every pool in the JVM stopped with it, and silently, which is the failure mode the hand-off of the close exists to avoid. The bound of the pool reaches the server log, throttled like the stall warning, rather than only the error the client is given. The javadoc of the reentrancy exemption said a nested borrow is never pooled on return. pollIdle runs before the reentrancy branch, so one served out of the idle deque carries the permit that connection already holds and is pooled like any other; only one that had to establish a connection of its own is not. The exemption is from the wait rather than from the pool. Seven cases added, none of which needs a database, and the two that back-date a connection are taken out of the reach of the sweeper running beside them. --- .../backends/jdbc/CachedConnection.java | 75 +++++-- .../server/backends/jdbc/JDBCStorage.java | 59 ++++-- .../jdbc/CachedConnectionTestCase.java | 200 +++++++++++++++++- 3 files changed, 300 insertions(+), 34 deletions(-) diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java index e08eca67f9..c743021ec9 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/CachedConnection.java @@ -66,6 +66,7 @@ public class CachedConnection implements Connection { private static final Executor DIRECT_EXECUTOR = Runnable::run; private static final AtomicLong lastStallWarning = new AtomicLong(); + private static final AtomicLong lastPoolFullWarning = new AtomicLong(); private static final AtomicBoolean readBoundWarned = new AtomicBoolean(); final Connection parent; @@ -138,14 +139,17 @@ private static void startSweeper() { // Expiry has to happen without a borrow behind it. Caffeine was left without a scheduler, so an // entry was only ever expired by a later cache operation - and a backend that has gone idle, // the one case the TTL exists for, performs none (issue #878). - private static void sweep() { + static void sweep() { final long ttlMillis = getCacheTtlMillis(); final Executor closeOn = closer; for (final Pool pool : pools.values()) { try { pool.sweep(ttlMillis, closeOn); - } catch (RuntimeException e) { - logger.traceException(e); + } catch (Throwable t) { + // Error included: scheduleWithFixedDelay cancels a task that throws, so anything + // escaping here would stop the expiry of every pool in the JVM for good - and + // silently, which is the failure mode the hand-off of the close exists to avoid. + logger.traceException(t); } } } @@ -198,8 +202,12 @@ static final class Pool { * for the first to be returned would wait for this very thread: * {@code PersistentCompressedSchema.store()} opens a write of its own - the definition has * to commit independently of the entry - and {@code EntryContainer.modifyDN} reaches it - * from inside a transaction, having encoded the entry there. Such a connection is never - * pooled on return, so the pool does not grow past its bound. + * from inside a transaction, having encoded the entry there. The exemption is from the + * wait rather than from the pool: a nested borrow served out of the idle deque carries the + * permit that connection already holds and is pooled again on return like any other. Only + * one that had to establish a connection of its own, because the pool stood at its bound, + * holds no permit - and that one is closed rather than pooled when it comes back, so the + * pool does not grow past its bound. *

* Counted per pool rather than per thread, because that deadlock only exists within one * pool: a count shared by all of them would judge a thread holding a connection to one @@ -286,8 +294,14 @@ void drainIdle() { /** * Takes a connection out of the pool, waiting up to waitMs for one to be returned, and * discarding the ones that are broken or have been idle for longer than the TTL. + *

+ * Bounded by the deadline of the borrow, and not only by waitMs: a poll of no duration + * still hands out whatever the deque holds, and discarding a connection whose socket is + * half-open costs the validation timeout apiece. The pool holds as many of those as its + * bound allows, so draining the deque overran the bound the operator set - by minutes on a + * large pool, before the connect that follows it had even started (issue #878). */ - CachedConnection pollIdle(long waitMs, long ttlMillis) throws InterruptedException { + CachedConnection pollIdle(long waitMs, long ttlMillis, long deadline) throws InterruptedException { long remainingWait = waitMs; while (true) { final long polledAt = System.currentTimeMillis(); @@ -295,15 +309,16 @@ CachedConnection pollIdle(long waitMs, long ttlMillis) throws InterruptedExcepti if (con == null) { return null; } - if (System.currentTimeMillis() - con.returnedAtMillis <= ttlMillis && isUsable(con)) { + if (System.currentTimeMillis() - con.returnedAtMillis <= ttlMillis && isUsable(con, deadline)) { return con; } destroy(con); - remainingWait -= System.currentTimeMillis() - polledAt; - if (remainingWait <= 0) { - // one more look, since a connection may have been returned in the meantime - remainingWait = 0; + final long remaining = deadline - System.currentTimeMillis(); + if (remaining <= 0) { + return null; } + // one more look, since a connection may have been returned in the meantime + remainingWait = Math.min(Math.max(0, remainingWait - (System.currentTimeMillis() - polledAt)), remaining); } } @@ -365,9 +380,15 @@ void sweep(long ttlMillis, Executor closeOn) { // From the tail: the least recently returned connection is the first to have expired, // and once one has not, neither has anything in front of it. for (CachedConnection con = idle.peekLast(); con != null; con = idle.peekLast()) { - if (con.returnedAtMillis > deadline || !idle.removeLastOccurrence(con)) { + if (con.returnedAtMillis > deadline) { return; } + if (!idle.removeLastOccurrence(con)) { + // A borrow took it between the two. What is behind it may still have expired, + // and ending the cycle here would leave every one of those open until the + // next sweep. + continue; + } final CachedConnection expired = con; try { closeOn.execute(() -> destroy(expired)); @@ -557,7 +578,7 @@ static Connection getConnection(String connectionString) throws Exception { long backoffMs = 0; int attempts = 0; while (true) { - final CachedConnection pooled = pool.pollIdle(waitMs, ttlMillis); + final CachedConnection pooled = pool.pollIdle(waitMs, ttlMillis, deadline); if (pooled != null) { return borrowed(pooled); } @@ -568,9 +589,14 @@ static Connection getConnection(String connectionString) throws Exception { // and the only ceiling left would be the max_connections of the database itself. final long remaining = deadline - System.currentTimeMillis(); if (remaining <= 0) { - throw new SQLTimeoutException("no connection to " + safeUrl(connectionString) + final String message = "no connection to " + safeUrl(connectionString) + " could be borrowed within " + poolTimeoutSeconds + "s: all " + pool.max() - + " connections of the pool are in use (raise " + POOL_MAX_PROPERTY + " to allow more)"); + + " connections of the pool are in use (raise " + POOL_MAX_PROPERTY + " to allow more)"; + // The one failure the bound introduces has to reach the server log too: an + // installation whose peak sits above the default would otherwise see its + // operations fail with nothing in the log naming the pool behind it. + warnPoolFull(message); + throw new SQLTimeoutException(message); } waitMs = Math.min(POOL_FULL_POLL_MS, remaining); continue; @@ -618,12 +644,14 @@ static Connection getConnection(String connectionString) throws Exception { } } - private static boolean isUsable(CachedConnection con) { + private static boolean isUsable(CachedConnection con, long deadline) { try { // The validation needs a bound of its own: isValid(0) means "no timeout" in the JDBC // contract, and a connection whose socket is half-open answers it no sooner than it - // answers anything else. - return con.isValid(VALIDATION_TIMEOUT_SECONDS); + // answers anything else. Never longer than what is left of the borrow either, since + // that is the bound the caller was given - and never 0, which would lift it entirely. + final long remainingSeconds = (deadline - System.currentTimeMillis() + 999) / 1000; + return con.isValid((int) Math.max(1, Math.min(VALIDATION_TIMEOUT_SECONDS, remainingSeconds))); } catch (SQLException e) { // a driver reporting the validation as an error: discard it return false; } @@ -685,6 +713,17 @@ static boolean isConnectionLimit(SQLException e, ConnectDialect dialect) { return false; } + // The bound of the pool is a reason for an operation to fail that no version before it had, + // so it belongs in the server log as well as in the error the client is given. Throttled like + // the stall warning: every worker thread reaches it at once when the pool stands full. + private static void warnPoolFull(String message) { + final long now = System.currentTimeMillis(); + final long last = lastPoolFullWarning.get(); + if (now - last >= STALL_WARNING_INTERVAL_MS && lastPoolFullWarning.compareAndSet(last, now)) { + logger.warn(LocalizableMessage.raw("%s", message)); + } + } + // A stall has to reach the server log: without it a database accepting no further connection // is indistinguishable from a hang. Throttled, since every operation of the backend borrows // through here and would otherwise log a copy of its own. diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java index 58eebdb5e8..ac769aaeb4 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java @@ -141,7 +141,16 @@ void executeAny(PreparedStatement statement) throws SQLException { } Connection getConnection() throws Exception { - return CachedConnection.getConnection(config.getDBDirectory()); + // The pool this storage registered with in open(), not the one config names now. Nothing + // keeps db-directory from being changed on a running backend - applyConfigurationChange() + // takes it, isConfigurationChangeAcceptable() refuses nothing, and the component-restart + // admin action renders a message rather than holding the change back - so re-reading it + // here would borrow from a pool this storage never registered with, leaving the one it did + // register with holding a user that never borrows: the leak of #878 back through the + // configuration. And an unregistered pool is drained the moment another backend that did + // register with it closes, with this one still borrowing from it (issue #878). + final String registered=poolConnectionString; + return CachedConnection.getConnection(registered!=null ? registered : config.getDBDirectory()); } @@ -1577,6 +1586,27 @@ public void aborted() { aborted = true; } + /** + * Hands the connection back to the pool and closes the stamp session, whatever went before. + * Returns the failure the caller is to report: the return rolls back, and the rollback + * fails on exactly the connection whose commit just did, so the commit stays the exception + * the caller sees and this one rides along with it instead of replacing it. + */ + private SQLException releaseConnection(SQLException failure) { + try { + con.close(); + } catch (SQLException e) { + if (failure==null) { + failure=e; + }else { + failure.addSuppressed(e); + } + } finally { + txw.stampSession.close(); + } + return failure; + } + @Override public void close() { try { @@ -1590,23 +1620,22 @@ public void close() { } } catch (SQLException e) { failure=e; + } catch (Throwable t) { + // Back to the pool whatever came out of the commit, not only on the SQLException + // a driver is supposed to throw: nothing else holds this connection, and only + // its close() gives back the permit it took. A pool is never removed from the + // map, so a permit lost to an Error out of a bulk import - or to a driver + // failing unchecked - is lost for the life of the server, and enough of them + // walk the bound down to nothing (issue #878). + final SQLException onTheWayOut=releaseConnection(null); + if (onTheWayOut!=null) { + t.addSuppressed(onTheWayOut); + } + throw t; } // Back to the pool even when the commit failed: nothing else holds this connection, // so leaving it behind would leak it along with the failure. - try { - con.close(); - } catch (SQLException e) { - // The return rolls back, and the rollback fails on exactly the connection whose - // commit just did. The commit is what went wrong, so it stays the exception the - // caller sees and this one rides along with it instead of replacing it. - if (failure==null) { - failure=e; - }else { - failure.addSuppressed(e); - } - } finally { - txw.stampSession.close(); - } + failure=releaseConnection(failure); if (failure!=null) { throw new StorageRuntimeException(failure); } diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java index 49e2ff90df..1184c3c886 100644 --- a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java +++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/CachedConnectionTestCase.java @@ -15,7 +15,10 @@ */ package org.opends.server.backends.jdbc; +import org.forgerock.opendj.server.config.server.JDBCBackendCfg; import org.opends.server.DirectoryServerTestCase; +import org.opends.server.backends.pluggable.spi.AccessMode; +import org.opends.server.backends.pluggable.spi.Importer; import org.testng.annotations.AfterClass; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeClass; @@ -37,9 +40,11 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.logging.Logger; import static org.mockito.Mockito.anyInt; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -126,11 +131,16 @@ public void testThePoolDoesNotGrowPastItsBound() throws Exception { /** Borrows the way the server does, one operation to a thread. */ private static Connection borrowOnAThreadOfItsOwn(String url) throws Exception { + return startBorrow(url).get(120, TimeUnit.SECONDS); + } + + /** The same, left running: a borrow that waits has to be looked at while it does. */ + private static FutureTask startBorrow(String url) { final FutureTask borrow = new FutureTask<>(() -> CachedConnection.getConnection(url)); final Thread thread = new Thread(borrow, "borrow-" + url); thread.setDaemon(true); thread.start(); - return borrow.get(120, TimeUnit.SECONDS); + return borrow; } /** @@ -196,6 +206,9 @@ public void testTheSweepClosesAnIdleConnectionWithNoBorrowBehindIt() throws Exce stub.answerWith(null); final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url); con.close(); + // The sweeper of the server is running while this case does, over every pool and reading + // the TTL as it goes: out of its reach, so that the sweep asserted here is the one below. + System.setProperty(CachedConnection.TTL_PROPERTY, "600000"); con.returnedAtMillis = System.currentTimeMillis() - 60000; final CachedConnection.Pool pool = CachedConnection.poolOf(url); assertEquals(pool.idleCount(), 1); @@ -358,6 +371,7 @@ public void testTheSweepDoesNotCloseOnTheSweeperThread() throws Exception { stub.answerWith(null); final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url); con.close(); + System.setProperty(CachedConnection.TTL_PROPERTY, "600000"); // see the case above con.returnedAtMillis = System.currentTimeMillis() - 60000; final CachedConnection.Pool pool = CachedConnection.poolOf(url); final List handedOff = new ArrayList<>(); @@ -373,6 +387,190 @@ public void testTheSweepDoesNotCloseOnTheSweeperThread() throws Exception { assertEquals(pool.liveCount(), 0, "a swept connection kept its permit"); } + /** + * And the sweep the scheduled sweeper actually runs closes elsewhere too: the case above + * supplies an executor of its own, so it would pass just as well with the production one left + * closing inline. + */ + @Test(timeOut = 120000) + public void testTheScheduledSweepClosesOnAThreadOfItsOwn() throws Exception { + final String url = StubDriver.PREFIX + "sweeper-thread"; + stub.answerWith(null); + final CachedConnection con = (CachedConnection) CachedConnection.getConnection(url); + con.close(); + final AtomicReference closedOn = new AtomicReference<>(); + doAnswer(invocation -> { + closedOn.set(Thread.currentThread().getName()); + return null; + }).when(con.parent).close(); + con.returnedAtMillis = System.currentTimeMillis() - 60000; + System.setProperty(CachedConnection.TTL_PROPERTY, "1000"); + + CachedConnection.sweep(); // what the scheduled sweeper runs, with nothing supplied to it + + for (int i = 0; i < 200 && closedOn.get() == null; i++) { + Thread.sleep(50); + } + assertNotNull(closedOn.get(), "the sweep never closed the expired connection"); + assertFalse(closedOn.get().contains("sweeper"), "the close ran on the sweeper thread: " + closedOn.get()); + assertTrue(closedOn.get().startsWith("JDBC backend connection pool closer"), closedOn.get()); + } + + /** + * A borrow may not outlast the deadline it was given while emptying the pool. A poll of no + * duration still hands out whatever the deque holds, and a connection whose socket is half-open + * - a moved VIP, a firewall that dropped the idle sockets - costs the validation timeout to + * discard, so draining a pool of its full bound overran the deadline by minutes, before the + * connect that follows it had even started (#878). + */ + @Test(timeOut = 120000) + public void testABorrowStopsAtItsDeadlineRatherThanDrainingThePool() throws Exception { + final String url = StubDriver.PREFIX + "deadline-drain"; + System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "6"); + System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "1"); + final CachedConnection.Pool pool = CachedConnection.poolOf(url); + + // Fresh by the TTL, and each one a second to find broken: the pool a burst of traffic left + // behind, against a database that has stopped answering. + for (int i = 0; i < 6; i++) { + final Connection halfOpen = mock(Connection.class); + when(halfOpen.isValid(anyInt())).thenAnswer(invocation -> { + Thread.sleep(1000); + return false; + }); + assertTrue(pool.tryReserve()); + pool.addIdle(new CachedConnection(url, halfOpen, pool, true)); + } + stub.answerWith(null); + + final long startedAt = System.currentTimeMillis(); + final Connection borrowed = CachedConnection.getConnection(url); + final long elapsed = System.currentTimeMillis() - startedAt; + + assertNotNull(borrowed); + assertTrue(elapsed < 3500, "the borrow drained the pool past its deadline: " + elapsed + " ms"); + borrowed.close(); + CachedConnection.invalidate(url); + } + + /** 0 means "no bound" for the size of the pool, and an invalid value means "the default". */ + @Test(timeOut = 120000) + public void testTheBoundOfThePoolReadsItsBoundaryValues() throws Exception { + assertEquals(poolWithMax("unbounded", "0").max(), Integer.MAX_VALUE, "0 must mean no bound"); + assertEquals(poolWithMax("negative", "-1").max(), CachedConnection.DEFAULT_POOL_MAX); + assertEquals(poolWithMax("not-a-number", "sixteen").max(), CachedConnection.DEFAULT_POOL_MAX); + } + + private static CachedConnection.Pool poolWithMax(String name, String max) { + System.setProperty(CachedConnection.POOL_MAX_PROPERTY, max); + return CachedConnection.poolOf(StubDriver.PREFIX + "bound-" + name); // read when the pool is built + } + + /** 0 means "wait without limit" for a borrow, rather than "give up at once". */ + @Test(timeOut = 120000) + public void testABorrowWithNoDeadlineWaitsForAReturnedConnection() throws Exception { + final String url = StubDriver.PREFIX + "no-deadline"; + System.setProperty(CachedConnection.POOL_MAX_PROPERTY, "1"); + System.setProperty(CachedConnection.POOL_TIMEOUT_PROPERTY, "0"); + stub.answerWith(null); + + final Connection held = borrowOnAThreadOfItsOwn(url); + final FutureTask waiting = startBorrow(url); + try { + waiting.get(1500, TimeUnit.MILLISECONDS); + fail("the borrow gave up although it was given no deadline"); + } catch (TimeoutException expected) { + // still waiting for the connection of the pool to come back, which is the point + } + + held.close(); + final Connection served = waiting.get(120, TimeUnit.SECONDS); + assertSame(served, held, "the borrow was served by something other than the returned connection"); + served.close(); + CachedConnection.invalidate(url); + } + + /** 0 means "keep nothing" for the TTL: an idle connection is not handed out again. */ + @Test(timeOut = 120000) + public void testAZeroTtlKeepsNoIdleConnection() throws Exception { + final String url = StubDriver.PREFIX + "zero-ttl"; + stub.answerWith(null); + final CachedConnection first = (CachedConnection) CachedConnection.getConnection(url); + first.close(); + first.returnedAtMillis = System.currentTimeMillis() - 5; + System.setProperty(CachedConnection.TTL_PROPERTY, "0"); + + final Connection second = CachedConnection.getConnection(url); + + assertNotSame(second, first, "a connection was kept although the TTL keeps none"); + verify(first.parent).close(); + second.close(); + CachedConnection.invalidate(url); + } + + /** + * The storage borrows from the pool it registered with, and gives that registration back when + * it closes. db-directory may be changed on a running backend - applyConfigurationChange takes + * it and nothing refuses it - and a borrow that followed the change would leave the pool this + * storage registered with holding a user that never borrows, while the pool it borrowed from + * has none: the leak of #878 back through the configuration, and a pool another backend may + * drain while this one is still borrowing from it. + */ + @Test(timeOut = 120000) + public void testTheStorageBorrowsFromThePoolItRegisteredWith() throws Exception { + final String registered = StubDriver.PREFIX + "storage-registered"; + final String changed = StubDriver.PREFIX + "storage-changed"; + stub.answerWith(null); + final JDBCBackendCfg cfg = mock(JDBCBackendCfg.class); + when(cfg.getDBDirectory()).thenReturn(registered); + final JDBCStorage storage = new JDBCStorage(cfg, null); + storage.open(AccessMode.READ_WRITE); + + when(cfg.getDBDirectory()).thenReturn(changed); // the configuration changed under it + try (final Connection con = storage.getConnection()) { + assertEquals(((CachedConnection) con).connectionString, registered, + "the borrow left the pool this storage registered with"); + } + assertEquals(CachedConnection.poolOf(changed).liveCount(), 0, "a pool with no user was borrowed from"); + + storage.close(); + + assertEquals(CachedConnection.poolOf(registered).idleCount(), 0, + "close() left the connections of the pool it registered with behind"); + } + + /** + * An import gives its connection back however its commit went. The commit used to be guarded + * against SQLException alone, so an Error out of a bulk import - or a driver failing unchecked + * - left the connection borrowed and its permit with it; a pool is never removed from the map, + * so that permit was gone for the life of the server and enough imports walked the bound of + * the pool down to nothing (#878). + */ + @Test(timeOut = 120000) + public void testAnImportGivesItsConnectionBackWhenTheCommitFailsUnchecked() throws Exception { + final String url = StubDriver.PREFIX + "import-unchecked-commit"; + final Connection parent = mock(Connection.class); + when(parent.isValid(anyInt())).thenReturn(true); + doThrow(new Error("out of memory while importing")).when(parent).commit(); + stub.answerWith(parent); + final JDBCBackendCfg cfg = mock(JDBCBackendCfg.class); + when(cfg.getDBDirectory()).thenReturn(url); + final JDBCStorage storage = new JDBCStorage(cfg, null); + storage.open(AccessMode.READ_WRITE); + final Importer importer = storage.startImport(); + + try { + importer.close(); + fail("the failure of the commit was not reported"); + } catch (Error expected) { + // reported to the caller, which is what an Error out of an import has to be + } + + assertEquals(CachedConnection.poolOf(url).idleCount(), 1, "the import kept the connection of the pool"); + storage.close(); + assertEquals(CachedConnection.poolOf(url).liveCount(), 0, "the import kept a permit of the pool"); + } + /** * A driver that is not on the classpath - the JDBC backend needs one dropped into * lib/extensions by hand - is a configuration error the caller has to see. Retried, it is