This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit b86ff4d9905b8dd63688565b5f92308e42401180 Author: opencode <[email protected]> AuthorDate: Tue Oct 27 21:47:37 2026 +0100 Bind cached statements to the connection they were prepared on The statement cache relied entirely on the disconnected() purge hook having run before the physical connection was replaced. The reported route into that assumption, an unguarded disconnectEvent() walk that a throwing interceptor could stop before the cache was reached, had already been closed by the earlier interceptor isolation work, but a live window remained: a statement close racing a disconnect/reconnect can park an open statement of the old connection in the cache after the purge had already run, and the next borrower's cache hit then serves a statement of a dead connection (reproduced against H2 as "The object is already closed"). An event skipped behind a failed connect has the same effect. A cache-time generation stamp cannot cover this, since the racing insert lands under the new stamp. Each cached statement now carries the physical connection it was prepared on, as reported by the driver for the statement itself, with the pool's current connection as the fallback when the driver will not answer. cacheStatement() refuses entries whose connection is no longer the current one, before touching the pool-wide counter, and a cache hit on such an entry closes it and prepares the statement again on the current connection instead of handing it to the borrower. Add TestStatementCacheRacingClose, which deterministically reproduces the racing close through production code paths and fails on the previous code, TestStatementCacheStaleEntryResurrection, which pins the hit-path guard, and TestStatementCacheReconnect, which pins the return-time, borrow-time and sweeper reconnect purges behind an interceptor that throws from disconnected(). The throwing interceptor used by these tests moves to the shared test base. --- .../jdbc/pool/interceptor/StatementCache.java | 62 +++++ .../org/apache/tomcat/jdbc/JdbcPoolBaseTest.java | 21 ++ .../pool/TestThrowingDisconnectedInterceptor.java | 13 -- .../interceptor/TestStatementCacheRacingClose.java | 175 ++++++++++++++ .../interceptor/TestStatementCacheReconnect.java | 260 +++++++++++++++++++++ .../TestStatementCacheStaleEntryResurrection.java | 138 +++++++++++ 6 files changed, 656 insertions(+), 13 deletions(-) diff --git a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/interceptor/StatementCache.java b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/interceptor/StatementCache.java index c50016c413..c40226810b 100644 --- a/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/interceptor/StatementCache.java +++ b/modules/jdbc-pool/src/main/java/org/apache/tomcat/jdbc/pool/interceptor/StatementCache.java @@ -18,6 +18,7 @@ package org.apache.tomcat.jdbc.pool.interceptor; import java.lang.reflect.Constructor; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; +import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; @@ -249,6 +250,15 @@ public class StatementCache extends StatementDecoratorInterceptor implements Sta if (statement!=null) { //remove it from the cache since it is used removeStatement(statement); + if (statement.getAssociatedConnection() != currentPhysicalConnection()) { + // The entry belongs to a connection that has been + // replaced (a statement close that raced a reconnect + // parked it after the disconnect purge ran). It is a + // statement of a dead connection: close it and prepare + // again rather than hand it to this borrower. + closeStatement(statement); + return super.invoke(proxy, method, args); + } // The cached proxy carries the connection handle of the // borrower that closed the statement, which by now is a // disposed facade. Re-bind it to the borrower the proxy is @@ -296,6 +306,13 @@ public class StatementCache extends StatementDecoratorInterceptor implements Sta } if (proxy.getCacheKey()==null) { return false; + } else if (proxy.getAssociatedConnection() != currentPhysicalConnection()) { + // The physical connection the statement was prepared on has + // been replaced (typically a close that raced a + // disconnect/reconnect, landing after the disconnected() purge + // had already run). The statement belongs to a dead connection + // and must not be parked in the cache of the new one. + return false; } else if (cache.containsKey(proxy.getCacheKey())) { return false; } else if (cacheSize.get()>=maxCacheSize) { @@ -330,6 +347,17 @@ public class StatementCache extends StatementDecoratorInterceptor implements Sta } /*end the actual statement cache*/ + /** + * Returns the physical connection behind the pooled connection this + * interceptor is bound to, or {@code null} if there is none (yet). + * + * @return the current physical connection, or null + */ + private Connection currentPhysicalConnection() { + PooledConnection pCon = this.pcon; + return pCon == null ? null : pCon.getConnection(); + } + /** * Returns the statement cache for the current pooled connection. * @@ -366,6 +394,13 @@ public class StatementCache extends StatementDecoratorInterceptor implements Sta * Cache key for this statement. */ CacheKey key; + /** + * The physical connection this statement was prepared on. Cached + * statements are bound to the connection they were prepared on; if + * the pooled connection is reconnected, the entries become stale + * and must neither be cached nor handed out. + */ + private final Connection associatedConnection; /** * Poolable hint set by the application. Tracked here as some drivers * (e.g. H2) always return false from {@code isPoolable()}. @@ -378,6 +413,33 @@ public class StatementCache extends StatementDecoratorInterceptor implements Sta */ public CachedStatement(PreparedStatement parent, String sql) { super(parent, sql); + associatedConnection = associatedConnectionOf(parent); + } + + /** + * Resolves the physical connection a statement belongs to. The + * connection the driver reports for the statement itself is + * authoritative; only when the driver will not answer does the + * pool's own record of the current connection serve as the binding. + * + * @param statement the statement whose associated connection to + * determine + * @return the physical connection of the statement + */ + private Connection associatedConnectionOf(PreparedStatement statement) { + try { + return statement.getConnection(); + } catch (SQLException x) { + return currentPhysicalConnection(); + } + } + + /** + * Returns the physical connection this statement was prepared on. + * @return the connection associated with the statement + */ + public Connection getAssociatedConnection() { + return associatedConnection; } @Override diff --git a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/JdbcPoolBaseTest.java b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/JdbcPoolBaseTest.java index f5791cede3..d090336cff 100644 --- a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/JdbcPoolBaseTest.java +++ b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/JdbcPoolBaseTest.java @@ -31,8 +31,10 @@ import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeAll; +import org.apache.tomcat.jdbc.pool.ConnectionPool; import org.apache.tomcat.jdbc.pool.DataSource; import org.apache.tomcat.jdbc.pool.JdbcInterceptor; +import org.apache.tomcat.jdbc.pool.PooledConnection; import org.h2.tools.Server; /** @@ -162,6 +164,25 @@ public abstract class JdbcPoolBaseTest { return null; } + /* + * An interceptor that always throws from disconnected(), shared by the + * tests that pin how the disconnect path isolates a misbehaving link of + * the chain. It is configured through the jdbcInterceptors class-name + * list, so it must stay a public static class with a no-arg constructor. + */ + public static class BoomInterceptor extends JdbcInterceptor { + @Override + public void reset(ConnectionPool parent, PooledConnection con) { + // empty + } + + @Override + public void disconnected(ConnectionPool parent, PooledConnection con, + boolean finalizing) { + throw new IllegalStateException("boom from disconnected()"); + } + } + /* * The driver's own object behind a pooled connection's statement. For * a prepared statement this is the honest oracle for "the session was diff --git a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestThrowingDisconnectedInterceptor.java b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestThrowingDisconnectedInterceptor.java index db419b1098..13c4388ded 100644 --- a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestThrowingDisconnectedInterceptor.java +++ b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/TestThrowingDisconnectedInterceptor.java @@ -33,19 +33,6 @@ import org.apache.tomcat.jdbc.JdbcPoolBaseTest; */ class TestThrowingDisconnectedInterceptor extends JdbcPoolBaseTest { - public static class BoomInterceptor extends JdbcInterceptor { - @Override - public void reset(ConnectionPool parent, PooledConnection con) { - // empty - } - - @Override - public void disconnected(ConnectionPool parent, PooledConnection con, - boolean finalizing) { - throw new IllegalStateException("boom from disconnected()"); - } - } - public static class RecordingInterceptor extends JdbcInterceptor { static final AtomicInteger DISCONNECTED_CALLS = new AtomicInteger(); diff --git a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementCacheRacingClose.java b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementCacheRacingClose.java new file mode 100644 index 0000000000..1b781f475f --- /dev/null +++ b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementCacheRacingClose.java @@ -0,0 +1,175 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.tomcat.jdbc.pool.interceptor; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.sql.Statement; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import org.apache.tomcat.jdbc.JdbcPoolBaseTest; +import org.apache.tomcat.jdbc.pool.DataSource; +import org.apache.tomcat.jdbc.pool.PooledConnection; + +/** + * A statement close racing a session replacement must not leave the closed + * statement of the old session parked in the cache of the new one. The + * close cleans up the live delegate first and only then inserts into the + * map; a disconnect/reconnect can land in between, so the purge of the old + * session has already run while the map insert of the dead session's + * statement is still pending. Without a session check on the way in, that + * insert parks a dead-session entry and the next borrower's cache hit is + * served its statement. + * + * <p>The race is emulated deterministically with production code only: the + * cache is configured as a subclass whose {@code cacheStatement} override + * performs the reconnect first and runs the real caching afterwards - + * exactly the interleaving the racing close and the racing reconnect take: + * cleanup on the live delegate done, session replaced, map insert pending. + * Pre-fix (no session check in {@code cacheStatement}) the override let + * that insert succeed, and the next borrower's cache hit returned the same + * driver statement of the dead session, whose every call answers an H2 + * "The object is already closed [90007-220]"; post-fix the entry is + * refused, the closing statement is closed normally, the cache stays empty + * and the next borrower gets a fresh working prepare on the new session.</p> + */ +class TestStatementCacheRacingClose extends JdbcPoolBaseTest { + + /** + * The statement cache with one hook: while armed, the next call to + * {@code cacheStatement} first replaces the physical session of the + * pooled connection - the window a racing disconnect/reconnect uses to + * land between the close's cleanup and its map insert. The pool builds + * interceptors reflectively from the configured class name, so this is + * a public static class with a no-arg constructor. + */ + public static class RacingCloseStatementCache extends StatementCache { + + private PooledConnection raceReconnectOnNextPark; + + /** + * Arms the hook: the next statement parked through + * {@code cacheStatement} meets a replacement of this pooled + * connection's session inside that call. + * + * @param pcon the pooled connection whose session is to be replaced + */ + void raceNextParkWithReconnect(PooledConnection pcon) { + raceReconnectOnNextPark = pcon; + } + + @Override + public boolean cacheStatement(CachedStatement proxy) { + PooledConnection raced = raceReconnectOnNextPark; + if (raced != null) { + raceReconnectOnNextPark = null; + try { + raced.reconnect(); + } catch (SQLException x) { + throw new IllegalStateException("the racing reconnect failed", x); + } + } + return super.cacheStatement(proxy); + } + } + + @Test + void aStatementClosedWhileTheSessionIsReplacedIsNotParkedInTheNewSessionsCache() + throws Exception { + DataSource ds = newDataSource(); + ds.getPoolProperties().setJdbcInterceptors( + RacingCloseStatementCache.class.getName() + "(prepared=true,max=5)"); + ds.getPoolProperties().setMaxActive(1); + // keep the background sweeper out of the way: the racing close + // performs the reconnect itself here + ds.getPoolProperties().setTimeBetweenEvictionRunsMillis(60000); + String oldTable = uniqueTable(); + String racingTable = uniqueTable(); + ds.createPool(); + try (Connection admin = adminConnection(); Statement st = admin.createStatement()) { + st.execute("CREATE TABLE " + oldTable + " (v INT)"); + st.execute("CREATE TABLE " + racingTable + " (v INT)"); + } + String parkedSql = "INSERT INTO " + oldTable + " VALUES (?)"; + String racingSql = "INSERT INTO " + racingTable + " VALUES (?)"; + + Connection first = ds.getConnection(); + PooledConnection pcon = (PooledConnection) first.unwrap(PooledConnection.class); + RacingCloseStatementCache cache = + interceptorOf(first, RacingCloseStatementCache.class); + setSessionMarker(first, 31); + + // one entry parked normally, so the purge the racing close runs + // into has real work to do + PreparedStatement parked = first.prepareStatement(parkedSql); + parked.setInt(1, 1); + Assertions.assertEquals(1, parked.executeUpdate()); + parked.close(); + Assertions.assertEquals(1, cache.getCacheSizePerConnection(), + "the statement must be waiting in the cache"); + + // the statement whose close races the session replacement + Connection oldPhysical = pcon.getConnection(); + PreparedStatement racing = first.prepareStatement(racingSql); + Statement racingDriverStatement = driverStatement(racing); + racing.setInt(1, 2); + Assertions.assertEquals(1, racing.executeUpdate()); + + cache.raceNextParkWithReconnect(pcon); + // on the live delegate the close's cleanup has just run; the + // session is replaced inside cacheStatement, before the map insert + racing.close(); + + Assertions.assertTrue(oldPhysical.isClosed(), + "the racing reconnect must have replaced the physical session"); + awaitSessions(1); + Assertions.assertEquals(0, cache.getCacheSizePerConnection(), + "cacheStatement must refuse the closed statement of the " + + "replaced session (pre-fix it parked it here, " + + "after the purge had already run)"); + Assertions.assertTrue(racingDriverStatement.isClosed(), + "the refused statement must have been closed by its close"); + + first.close(); // back to the pool on the new session + Connection second = ds.getConnection(); + try { + Assertions.assertNull(sessionMarker(second), + "the next borrower must sit on the new physical session"); + PreparedStatement reused = second.prepareStatement(racingSql); + // Pre-fix this is the cache hit on the dead-session entry: the + // same driver statement is served again, and every call on it + // answers the H2 error for a statement of a dead session. + Assertions.assertNotSame(racingDriverStatement, driverStatement(reused), + "the statement of the replaced session must not be served"); + reused.setInt(1, 3); + Assertions.assertEquals(1, reused.executeUpdate(), + "the next borrower must get a statement that works on its session"); + reused.close(); + Assertions.assertEquals(1, cache.getCacheSizePerConnection(), + "the new session must be able to park its own statement"); + } finally { + second.close(); + } + Assertions.assertEquals(1, adminRowCount(oldTable), + "the parked statement's execute must have reached the database"); + Assertions.assertEquals(2, adminRowCount(racingTable), + "both executes of the racing SQL must have reached the database"); + } +} diff --git a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementCacheReconnect.java b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementCacheReconnect.java new file mode 100644 index 0000000000..ade25c605d --- /dev/null +++ b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementCacheReconnect.java @@ -0,0 +1,260 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.tomcat.jdbc.pool.interceptor; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.Statement; +import java.util.Map; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import org.apache.tomcat.jdbc.JdbcPoolBaseTest; +import org.apache.tomcat.jdbc.pool.ConnectionPool; +import org.apache.tomcat.jdbc.pool.DataSource; +import org.apache.tomcat.jdbc.pool.JdbcInterceptor; +import org.apache.tomcat.jdbc.pool.PooledConnection; + +/** + * The statement cache is a cache of statements of one physical session: a + * cached statement is only valid while the session it was prepared on is + * the session the pooled connection is using. Every path that replaces the + * physical connection of a pooled connection - the reconnect of an expired + * session at return time, at borrow time, or from the idle sweeper - must + * therefore leave an empty cache behind, or the next borrower is handed a + * statement of a dead session as a cache hit. + * + * <p>The purge runs through the {@code disconnected()} callback, which a + * misbehaving interceptor can only survive because the disconnect event + * isolates every link of the chain (pinned generally by + * {@code TestThrowingDisconnectedInterceptor}). These tests pin the same + * guarantee for the statement cache specifically: an interceptor declared + * ahead of the cache that always throws from {@code disconnected()} must + * not keep the cache from being purged when the session is replaced.</p> + */ +class TestStatementCacheReconnect extends JdbcPoolBaseTest { + + /** + * Declared behind the cache; records the size of the statement cache as + * it is when the disconnect walk passes this link - i.e. after the + * cache itself has been notified. Records nothing at all when the walk + * never reaches behind the throwing link. + */ + public static class CacheAtDisconnectInterceptor extends JdbcInterceptor { + static volatile Integer cacheSizeAtDisconnect; + + @Override + public void reset(ConnectionPool parent, PooledConnection con) { + // empty + } + + @Override + @SuppressWarnings("rawtypes") + public void disconnected(ConnectionPool parent, PooledConnection con, + boolean finalizing) { + Object cache = con.getAttributes().get(StatementCache.STATEMENT_CACHE_ATTR); + cacheSizeAtDisconnect = + Integer.valueOf(cache == null ? 0 : ((Map) cache).size()); + } + } + + private DataSource cachePoolBehindThrower(int maxActive, long maxAgeMillis) { + DataSource ds = newDataSource(); + ds.getPoolProperties().setJdbcInterceptors( + BoomInterceptor.class.getName() + ";" + + StatementCache.class.getName() + "(prepared=true,max=5);" + + CacheAtDisconnectInterceptor.class.getName()); + ds.getPoolProperties().setMaxActive(maxActive); + ds.getPoolProperties().setMaxAge(maxAgeMillis); + // keep the background sweeper out of the way: these tests trigger + // the reconnect paths themselves (return, borrow, explicit + // testAllIdle) and pin the reconnect count exactly + ds.getPoolProperties().setTimeBetweenEvictionRunsMillis(60000); + return ds; + } + + /** The statement cache as the given borrower sees it. */ + private static StatementCache cacheOf(Connection pooled) { + return interceptorOf(pooled, StatementCache.class); + } + + /** + * What every physical-connection replacement must leave behind: no + * cached statement of the replaced session, a closed purge of it, and + * a fresh prepare for the next borrower. The assertions are true + * regardless of which path did the replacing, so the helper pins the + * invariant, not one call site. + */ + private void assertCacheDoesNotCrossSessions(DataSource ds, Connection second, + java.sql.Connection oldPhysical, Statement oldStatement, String insert, + int expectedInserts) throws Exception { + Assertions.assertNull(sessionMarker(second), + "the next borrower must sit on a new physical session"); + Assertions.assertEquals(0, cacheOf(second).getCacheSizePerConnection(), + "the cache must not carry a statement across a replaced session"); + Assertions.assertTrue(oldPhysical.isClosed(), + "the replaced physical session must be closed by the reconnect"); + Assertions.assertTrue(oldStatement.isClosed(), + "the purge must close the statement of the replaced session"); + awaitSessions(1); + PreparedStatement fresh = second.prepareStatement(insert); + Assertions.assertNotSame(oldStatement, driverStatement(fresh), + "the next borrower must get a fresh prepare, not the statement " + + "of the dead session"); + fresh.setInt(1, expectedInserts); + Assertions.assertEquals(1, fresh.executeUpdate()); + fresh.close(); + Assertions.assertEquals(1, cacheOf(second).getCacheSizePerConnection(), + "the new session must be able to cache its own statement"); + } + + @Test + void aThrowingInterceptorAheadOfTheCacheDoesNotSurviveTheReturnTimeReconnect() + throws Exception { + CacheAtDisconnectInterceptor.cacheSizeAtDisconnect = null; + DataSource ds = cachePoolBehindThrower(1, 1); + String table = uniqueTable(); + ds.createPool(); + try (Connection admin = adminConnection(); Statement st = admin.createStatement()) { + st.execute("CREATE TABLE " + table + " (v INT)"); + } + String insert = "INSERT INTO " + table + " VALUES (?)"; + + Connection first = ds.getConnection(); + java.sql.Connection oldPhysical = + ((PooledConnection) first.unwrap(PooledConnection.class)).getConnection(); + setSessionMarker(first, 11); + PreparedStatement prepared = first.prepareStatement(insert); + Statement oldStatement = driverStatement(prepared); + prepared.setInt(1, 1); + Assertions.assertEquals(1, prepared.executeUpdate()); + prepared.close(); + Assertions.assertEquals(1, cacheOf(first).getCacheSizePerConnection(), + "the statement must be waiting in the cache"); + + sleep(50); // the session is now unambiguously past its max age + first.close(); // the return reconnects it + + Assertions.assertEquals(Integer.valueOf(0), + CacheAtDisconnectInterceptor.cacheSizeAtDisconnect, + "the disconnect walk must carry on past the throwing link and " + + "reach the cache with the purge already done"); + Connection second = ds.getConnection(); + try { + assertCacheDoesNotCrossSessions(ds, second, oldPhysical, oldStatement, insert, 2); + } finally { + second.close(); + } + Assertions.assertEquals(2, adminRowCount(table), + "both executes must have reached the database"); + } + + @Test + void aThrowingInterceptorAheadOfTheCacheDoesNotSurviveTheBorrowTimeReconnect() + throws Exception { + CacheAtDisconnectInterceptor.cacheSizeAtDisconnect = null; + long maxAge = 300; + DataSource ds = cachePoolBehindThrower(1, maxAge); + String table = uniqueTable(); + ds.createPool(); + try (Connection admin = adminConnection(); Statement st = admin.createStatement()) { + st.execute("CREATE TABLE " + table + " (v INT)"); + } + String insert = "INSERT INTO " + table + " VALUES (?)"; + + Connection first = ds.getConnection(); + java.sql.Connection oldPhysical = + ((PooledConnection) first.unwrap(PooledConnection.class)).getConnection(); + setSessionMarker(first, 12); + PreparedStatement prepared = first.prepareStatement(insert); + Statement oldStatement = driverStatement(prepared); + prepared.setInt(1, 1); + Assertions.assertEquals(1, prepared.executeUpdate()); + prepared.close(); + Assertions.assertEquals(1, cacheOf(first).getCacheSizePerConnection(), + "the statement must be waiting in the cache"); + first.close(); // still within max age: goes back to the pool on its own session + long reconnectionsBeforeBorrow = ds.getPool().getReconnectedCount(); + + sleep(maxAge * 2); // the idle session expires + + Connection second = ds.getConnection(); + try { + Assertions.assertEquals(reconnectionsBeforeBorrow + 1, + ds.getPool().getReconnectedCount(), + "the borrow must be what reconnected the expired session"); + Assertions.assertEquals(Integer.valueOf(0), + CacheAtDisconnectInterceptor.cacheSizeAtDisconnect, + "the disconnect walk of the borrow-time reconnect must " + + "reach the cache with the purge already done"); + assertCacheDoesNotCrossSessions(ds, second, oldPhysical, oldStatement, insert, 2); + } finally { + second.close(); + } + Assertions.assertEquals(2, adminRowCount(table), + "both executes must have reached the database"); + } + + @Test + void aThrowingInterceptorAheadOfTheCacheDoesNotSurviveTheSweeperReconnect() + throws Exception { + CacheAtDisconnectInterceptor.cacheSizeAtDisconnect = null; + long maxAge = 300; + DataSource ds = cachePoolBehindThrower(1, maxAge); + String table = uniqueTable(); + ds.createPool(); + try (Connection admin = adminConnection(); Statement st = admin.createStatement()) { + st.execute("CREATE TABLE " + table + " (v INT)"); + } + String insert = "INSERT INTO " + table + " VALUES (?)"; + + Connection first = ds.getConnection(); + java.sql.Connection oldPhysical = + ((PooledConnection) first.unwrap(PooledConnection.class)).getConnection(); + setSessionMarker(first, 13); + PreparedStatement prepared = first.prepareStatement(insert); + Statement oldStatement = driverStatement(prepared); + prepared.setInt(1, 1); + Assertions.assertEquals(1, prepared.executeUpdate()); + prepared.close(); + Assertions.assertEquals(1, cacheOf(first).getCacheSizePerConnection(), + "the statement must be waiting in the cache"); + first.close(); // still within max age: goes back to the pool on its own session + long reconnectionsBeforeSweep = ds.getPool().getReconnectedCount(); + + sleep(maxAge * 2); // the idle session expires + + ds.getPool().testAllIdle(); // the idle sweeper's body reconnects it + + Assertions.assertEquals(reconnectionsBeforeSweep + 1, + ds.getPool().getReconnectedCount(), + "testAllIdle must have reconnected the expired session"); + Assertions.assertEquals(Integer.valueOf(0), + CacheAtDisconnectInterceptor.cacheSizeAtDisconnect, + "the disconnect walk of the sweeper reconnect must reach the " + + "cache with the purge already done"); + Connection second = ds.getConnection(); + try { + assertCacheDoesNotCrossSessions(ds, second, oldPhysical, oldStatement, insert, 2); + } finally { + second.close(); + } + Assertions.assertEquals(2, adminRowCount(table), + "both executes must have reached the database"); + } +} diff --git a/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementCacheStaleEntryResurrection.java b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementCacheStaleEntryResurrection.java new file mode 100644 index 0000000000..3f9770233a --- /dev/null +++ b/modules/jdbc-pool/src/test/java/org/apache/tomcat/jdbc/pool/interceptor/TestStatementCacheStaleEntryResurrection.java @@ -0,0 +1,138 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.tomcat.jdbc.pool.interceptor; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.Statement; +import java.util.Map; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import org.apache.tomcat.jdbc.JdbcPoolBaseTest; +import org.apache.tomcat.jdbc.pool.DataSource; +import org.apache.tomcat.jdbc.pool.PooledConnection; + +/** + * A cached statement belongs to the physical session it was prepared on, so + * the cache-hit path must never hand out an entry of a replaced session - + * even though the disconnect purge normally clears the cache when the pooled + * connection is reconnected, an entry of the replaced session can survive + * into the map behind that purge on production routes (a statement close + * racing the reconnect parks one after the purge ran; a failed connect keeps + * the next {@code disconnect(false)} from firing the purge event again, so + * an interim entry survives into the new session). + * + * <p>This class pins the cache-hit guard on its own, as defense-in-depth: + * the test injects a stale entry directly into the cache map after the + * reconnect completed - the very entry the purge had force-closed, a state + * production never re-parks by itself - and asserts the next borrower is not + * served that dead-session statement. The genuine racing close behind these + * entries is pinned faithfully by {@link TestStatementCacheRacingClose}. + * The injection manipulates the map directly, which production code never + * does and which leaves the pool-wide entry counter behind the map by + * design; that counter is therefore deliberately not asserted here, only + * the per-connection cache size.</p> + */ +class TestStatementCacheStaleEntryResurrection extends JdbcPoolBaseTest { + + @Test + @SuppressWarnings("rawtypes") + void anInjectedStaleEntryOfAReplacedSessionIsNotHandedToTheNextBorrower() throws Exception { + DataSource ds = newDataSource(); + ds.getPoolProperties().setJdbcInterceptors( + StatementCache.class.getName() + "(prepared=true,max=5)"); + ds.getPoolProperties().setMaxActive(1); + ds.getPoolProperties().setTimeBetweenEvictionRunsMillis(60000); + String table = uniqueTable(); + ds.createPool(); + try (Connection admin = adminConnection(); Statement st = admin.createStatement()) { + st.execute("CREATE TABLE " + table + " (v INT)"); + } + String insert = "INSERT INTO " + table + " VALUES (?)"; + + Connection first = ds.getConnection(); + java.sql.Connection oldPhysical = + ((PooledConnection) first.unwrap(PooledConnection.class)).getConnection(); + setSessionMarker(first, 21); + PreparedStatement prepared = first.prepareStatement(insert); + Statement oldDriverStatement = driverStatement(prepared); + prepared.setInt(1, 1); + Assertions.assertEquals(1, prepared.executeUpdate()); + prepared.close(); + + Map cacheMap = cacheOf(first).getCache(); + Assertions.assertEquals(1, cacheOf(first).getCacheSizePerConnection(), + "the statement must be waiting in the cache"); + Map.Entry poisoned = (Map.Entry) cacheMap.entrySet().iterator().next(); + Object poisonedKey = poisoned.getKey(); + Object poisonedEntry = poisoned.getValue(); + + // Replace the physical session underneath the pooled connection, as + // any reconnect path does: the purge runs and clears the map. + ((PooledConnection) first.unwrap(PooledConnection.class)).reconnect(); + Assertions.assertTrue(oldPhysical.isClosed(), + "the replaced session must be closed by the reconnect"); + Assertions.assertEquals(0, cacheMap.size(), + "the disconnect purge must have cleared the cache"); + awaitSessions(1); + + // Inject the stale entry directly: an entry of the replaced session - + // the very one the purge force-closed - sitting in the map while the + // pooled connection now sits on a new session. Production never + // re-parks a force-closed proxy; the hit-path guard must nonetheless + // refuse to hand it out. + cacheMap.put(poisonedKey, poisonedEntry); + + first.close(); // returned to the pool on the new session + Connection second = ds.getConnection(); + try { + Assertions.assertNull(sessionMarker(second), + "the next borrower must sit on the new physical session"); + Assertions.assertEquals(1, cacheOf(second).getCacheSizePerConnection(), + "the poisoned entry is what the next borrower takes over"); + + PreparedStatement reused = second.prepareStatement(insert); + // Pre-fix this hands out the proxy of the dead session: every + // call on it answers "Statement closed." instead of working. + Assertions.assertNotSame(oldDriverStatement, driverStatement(reused), + "the statement of the replaced session must not be served"); + Assertions.assertTrue(oldDriverStatement.isClosed(), + "the driver statement of the discarded entry must be closed"); + Assertions.assertEquals(0, cacheOf(second).getCacheSizePerConnection(), + "the discarded entry must be out of this connection's cache " + + "(the pool-wide counter lags the map by the " + + "injected re-add and is deliberately not asserted)"); + reused.setInt(1, 2); + Assertions.assertEquals(1, reused.executeUpdate(), + "the next borrower must get a statement that works on its session"); + reused.close(); + Assertions.assertEquals(1, cacheOf(second).getCacheSizePerConnection(), + "the new session caches its own statement"); + } finally { + second.close(); + } + Assertions.assertEquals(2, adminRowCount(table), + "both executes must have reached the database"); + } + + /** The statement cache as the given borrower sees it. */ + private static StatementCache cacheOf(Connection pooled) { + return interceptorOf(pooled, StatementCache.class); + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
