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 fcf3737df4..3ead059ad2 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 @@ -47,6 +47,27 @@ public class JDBCStorage implements org.opends.server.backends.pluggable.spi.Sto private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass(); + /** Number of attempts a {@link #write} makes before it propagates the conflict to the caller. */ + private static final int MAX_RETRIES = 10; + + /** Upper bound of the random delay before the second attempt, in milliseconds; it doubles with every attempt. */ + private static final double BASE_SLEEP_ON_RETRY_MS = 50.0; + + /** Upper bound the doubled delay is capped at, in milliseconds. */ + private static final double MAX_SLEEP_ON_RETRY_MS = 1000.0; + + /** Number of {@link Throwable#getCause()} hops walked when classifying a failure, also a guard against a cycle. */ + private static final int MAX_CAUSE_HOPS = 16; + + /** SQL Server error number of the transaction picked as the deadlock victim: "Rerun the transaction". */ + private static final int MSSQL_DEADLOCK_VICTIM = 1205; + + /** MySQL error number of a lock wait that timed out: ER_LOCK_WAIT_TIMEOUT, reported with SQLState HY000. */ + private static final int MYSQL_LOCK_WAIT_TIMEOUT = 1205; + + /** Oracle error number of a detected deadlock: ORA-00060, reported with SQLState 61000 rather than class 40. */ + private static final int ORACLE_DEADLOCK_DETECTED = 60; + private JDBCBackendCfg config; public JDBCStorage(JDBCBackendCfg cfg, ServerContext serverContext) { @@ -170,6 +191,18 @@ public void removeStorageFiles() throws StorageRuntimeException { } //operation + /** + * {@inheritDoc} + *

+ * A rolled back read is not replayed, although + * {@link org.opends.server.backends.pluggable.spi.Storage#read(ReadOperation)} asks for it: two of the read + * operations of this server are not idempotent, and replaying them corrupts their result rather than repairing + * it. {@code ExportJob} runs the whole export inside a single read and its LDIF writer is opened once, so a + * replay appends the entries already written instead of truncating the file; {@code VerifyJob} accumulates its + * counters in instance fields that no attempt resets, so a replay reports twice the entry count of the backend. + * Both are reachable while the server is online, since an export holds no more than a shared backend lock. + * A conflict therefore fails the read here, exactly as it did before the retry of {@link #write} was added. + */ @Override public T read(ReadOperation readOperation) throws Exception { try(final Connection con=getConnection()) { @@ -177,19 +210,115 @@ public T read(ReadOperation readOperation) throws Exception { } } + /** + * {@inheritDoc} + *

+ * {@link org.opends.server.backends.pluggable.spi.Storage#write(WriteOperation)} requires an implementation to + * retry a rolled back operation until it succeeds, and {@link WriteOperation} is documented as idempotent for + * exactly that reason; {@link org.opends.server.backends.pdb.PDBStorage#write(WriteOperation)} already does so + * on the conflict exception of its own engine. The loop is bounded here, unlike PDBStorage: the database may be + * shared with writers outside this server, so a conflict is not guaranteed to clear and failing the operation is + * better than never returning. + *

+ * Only the operation itself is replayed: a failure of {@link #getConnection()} or of the implicit + * {@link Connection#close()} - which returns the connection to the pool after a rollback - leaves the loop, so + * that a completed write is never replayed because releasing its connection failed. + */ @Override public void write(WriteOperation writeOperation) throws Exception { - try (final Connection con=getConnection()) { - try { - writeOperation.run(new WriteableTransactionTransactionImpl(con)); - con.commit(); - } catch (Exception e) { + for (int attempt=1;;attempt++) { + Exception failure=null; + String driver=null; + try (final Connection con=getConnection()) { + driver=getDriverName(con); try { - con.rollback(); - } catch (SQLException ex) {} - throw e; + writeOperation.run(new WriteableTransactionTransactionImpl(con)); + con.commit(); + return; + } catch (Exception e) { + try { + con.rollback(); + } catch (SQLException ex) {} + //rethrown, so that a failure of the implicit close() is suppressed into the failure being + //replayed rather than replacing it + failure=e; + throw e; + } + } catch (Exception e) { + //anything the operation did not throw comes from getConnection() or from the implicit close(), + //which returns the connection to the pool: neither belongs to the replayed region + if (e!=failure) { + throw e; + } + } + if (attempt>=MAX_RETRIES || !isRetryableConflict(failure,driver)) { + throw failure; + } + //logged rather than silently absorbed, so that a deployment retrying most of its writes stays observable + logger.warn(LocalizableMessage.raw("jdbc: replaying the transaction after a conflict, attempt %d of %d: %s", + attempt, MAX_RETRIES, stackTraceToSingleLineString(failure))); + try { + //randomized to spread the retries of the transactions that collided, growing to outlast contention + Thread.sleep(retryDelayMillis(attempt)); + } catch (InterruptedException e) { + //sleep cleared the interrupt flag: restore it, and report the failure being retried rather than the + //interrupt, which would hide from the caller what actually went wrong + Thread.currentThread().interrupt(); + failure.addSuppressed(e); + throw failure; + } + } + } + + /** Returns the randomized delay before the given attempt is replayed, doubling with each attempt up to a cap. */ + static long retryDelayMillis(int attempt) { + final double bound=Math.min(MAX_SLEEP_ON_RETRY_MS, BASE_SLEEP_ON_RETRY_MS * (1 << Math.min(attempt-1, 5))); + return (long) (Math.random() * bound); + } + + /** Returns the class name of the driver behind the given connection, which names the engine it talks to. */ + static String getDriverName(Connection con) { + return ((con instanceof CachedConnection) ? ((CachedConnection) con).parent : con).getClass().getName(); + } + + /** + * Returns whether the given failure carries a transaction conflict that replaying the operation can resolve. + *

+ * The conflict is looked up along the whole cause chain because it reaches this class wrapped: a deadlock in + * {@code put} arrives as {@code StorageRuntimeException(SQLException)}, and a caller such as + * {@code EntryContainer.addEntry} may wrap it once more. + *

+ * The standard class 40 states carry the conflict of most engines - 40P01 for PostgreSQL, 40001 for MySQL and + * for SQL Server - but not of all of them, so the vendor error numbers are consulted as well, keyed by the + * driver in the same way {@code getTableDialect} keys the column types. They cannot be matched + * driver-independently: Oracle reports a deadlock as ORA-00060 with SQLState 61000, and gives 1205 to a fatal + * "not a data file" error that no replay can resolve, while 1205 is exactly the deadlock victim of SQL Server + * and the lock wait timeout of MySQL. The MySQL timeout is not a deadlock, but it is transient in the same way + * and is equally resolved by a replay. The SQL Server number is matched beyond its class 40 state because a + * deployment may add {@code xopenStates=true} to its connection URL, which reports the same deadlock as 42000. + */ + static boolean isRetryableConflict(Throwable t, String driver) { + for (int hop=0; t!=null && hop + * The SQL Server driver sends {@link PreparedStatement#setString} parameters as NVARCHAR, and under a SQL + * collation comparing the {@code char(128)} column against an NVARCHAR value converts the column instead of + * the value: the primary key can no longer be sought, so every statement scans the whole table rather than + * reading one row. The upsert runs that scan under HOLDLOCK, which range-locks the entire table instead of + * the single key being written - the lock footprint that lets concurrent writers deadlock (error 1205). + * Casting the parameter back to char keeps the comparison seekable. + */ + static String hashParam(Connection con) { + return getDriverName(con).contains("microsoft") ? "cast(? as char(128))" : "?"; + } + private class ReadableTransactionImpl implements ReadableTransaction { final Connection con; boolean isReadOnly=true; @@ -229,7 +373,7 @@ public ReadableTransactionImpl(Connection con) { @Override public ByteString read(TreeName treeName, ByteSequence key) { - try (final PreparedStatement statement=con.prepareStatement("select v from "+getTableName(treeName)+" where h=? and k=?")){ + try (final PreparedStatement statement=con.prepareStatement("select v from "+getTableName(treeName)+" where h="+hashParam(con)+" and k=?")){ statement.setString(1,key2hash.get(ByteBuffer.wrap(key.toByteArray()))); statement.setBytes(2,real2db(key.toByteArray())); try(ResultSet rc=executeResultSet(statement)) { @@ -279,11 +423,11 @@ boolean isExistsTable(TreeName treeName) { } String getTableDialect() { - if (((CachedConnection) con).parent.getClass().getName().contains("oracle")) { + if (getDriverName(con).contains("oracle")) { return "h char(128),k raw(2000),v blob,primary key(h,k)"; - }else if (((CachedConnection) con).parent.getClass().getName().contains("mysql")) { + }else if (getDriverName(con).contains("mysql")) { return "h char(128),k varbinary(255),v longblob,primary key(h,k)"; - }else if (((CachedConnection) con).parent.getClass().getName().contains("microsoft")) { + }else if (getDriverName(con).contains("microsoft")) { return "h char(128),k varbinary(max),v image,primary key(h)"; } return "h char(128),k bytea,v bytea,primary key(h,k)"; @@ -301,7 +445,7 @@ public void openTree(TreeName treeName, boolean createOnDemand) { } } // CursorImpl iterates with "where k>? order by k" batches: primary key (h,k) cannot serve them - final String driverName=((CachedConnection) con).parent.getClass().getName(); + final String driverName=getDriverName(con); final String tableName=getTableName(treeName); if (driverName.contains("postgres")) { try (final PreparedStatement statement=con.prepareStatement("create index if not exists k_"+tableName.substring("opendj_".length())+" on "+tableName+" (k)")){ @@ -376,12 +520,15 @@ public void put(TreeName treeName, ByteSequence key, ByteSequence value) { try { upsert(treeName, key, value); } catch (SQLException e) { - throw new RuntimeException(e); + //StorageRuntimeException, like read() and delete(): EntryContainer passes that type through unchanged, + //while any other runtime exception is turned into an opaque ERR_UNCHECKED_EXCEPTION before it can be + //classified as a conflict + throw new StorageRuntimeException(e); } } boolean upsert(TreeName treeName, ByteSequence key, ByteSequence value) throws SQLException { - final String driverName=((CachedConnection) con).parent.getClass().getName(); + final String driverName=getDriverName(con); if (driverName.contains("postgres")) { //postgres upsert try (final PreparedStatement statement = con.prepareStatement("insert into " + getTableName(treeName) + " (h,k,v) values (?,?,?) ON CONFLICT (h, k) DO UPDATE set v=excluded.v")) { statement.setString(1, key2hash.get(ByteBuffer.wrap(key.toByteArray()))); @@ -403,8 +550,8 @@ boolean upsert(TreeName treeName, ByteSequence key, ByteSequence value) throws S statement.setBytes(3, value.toByteArray()); return (execute(statement) == 1 && statement.getUpdateCount() > 0); } - }else if (driverName.contains("microsoft")) { //ANSI MERGE with ; WITH (HOLDLOCK) makes the upsert atomic: without it SQL Server MERGE can race two concurrent NOT MATCHED inserts of the same key into a PRIMARY KEY violation. UPDLOCK is required on top of it: with HOLDLOCK alone the search phase takes a shared lock that the WHEN MATCHED update then has to convert to an exclusive one, so two concurrent upserts of the same key deadlock on the conversion; an update lock is taken right away and makes the second transaction wait instead - try (final PreparedStatement statement = con.prepareStatement("merge into " + getTableName(treeName) + " WITH (HOLDLOCK, UPDLOCK) old using (select ? h,? k,? v) new on (old.h=new.h and old.k=new.k) WHEN MATCHED THEN UPDATE SET old.v=new.v WHEN NOT MATCHED THEN INSERT (h,k,v) VALUES (new.h,new.k,new.v);")) { + }else if (driverName.contains("microsoft")) { //ANSI MERGE with ; WITH (HOLDLOCK) makes the upsert atomic: without it SQL Server MERGE can race two concurrent NOT MATCHED inserts of the same key into a PRIMARY KEY violation. UPDLOCK is required on top of it: with HOLDLOCK alone the search phase takes a shared lock that the WHEN MATCHED update then has to convert to an exclusive one, so two concurrent upserts of the same key deadlock on the conversion; an update lock is taken right away and makes the second transaction wait instead. h is cast back to char so that the join can seek the primary key instead of scanning the whole table under those locks, see hashParam() + try (final PreparedStatement statement = con.prepareStatement("merge into " + getTableName(treeName) + " WITH (HOLDLOCK, UPDLOCK) old using (select cast(? as char(128)) h,? k,? v) new on (old.h=new.h and old.k=new.k) WHEN MATCHED THEN UPDATE SET old.v=new.v WHEN NOT MATCHED THEN INSERT (h,k,v) VALUES (new.h,new.k,new.v);")) { statement.setString(1, key2hash.get(ByteBuffer.wrap(key.toByteArray()))); statement.setBytes(2, real2db(key.toByteArray())); statement.setBytes(3, value.toByteArray()); @@ -453,7 +600,7 @@ public boolean update(TreeName treeName, ByteSequence key, UpdateFunction f) { @Override public boolean delete(TreeName treeName, ByteSequence key) { - try (final PreparedStatement statement=con.prepareStatement("delete from "+getTableName(treeName)+" where h=? and k=?")){ + try (final PreparedStatement statement=con.prepareStatement("delete from "+getTableName(treeName)+" where h="+hashParam(con)+" and k=?")){ statement.setString(1,key2hash.get(ByteBuffer.wrap(key.toByteArray()))); statement.setBytes(2,real2db(key.toByteArray())); return (execute(statement)==1 && statement.getUpdateCount()>0); @@ -572,7 +719,7 @@ public void delete() throws NoSuchElementException, UnsupportedOperationExceptio if (isReadOnly) { throw new UnsupportedOperationException(); } - try (final PreparedStatement statement=con.prepareStatement("delete from "+tableName+" where h=? and k=?")){ + try (final PreparedStatement statement=con.prepareStatement("delete from "+tableName+" where h="+hashParam(con)+" and k=?")){ statement.setString(1,key2hash.get(ByteBuffer.wrap(db2real(currentKeyDb)))); statement.setBytes(2,currentKeyDb); execute(statement); @@ -616,7 +763,7 @@ && compareKeys(target,buffer.peekLast()[0])<=0) { @Override public boolean positionToKey(ByteSequence key) { final byte[] real=key.toByteArray(); - try (final PreparedStatement statement=con.prepareStatement("select v from "+tableName+" where h=? and k=?")){ + try (final PreparedStatement statement=con.prepareStatement("select v from "+tableName+" where h="+hashParam(con)+" and k=?")){ statement.setString(1,key2hash.get(ByteBuffer.wrap(real))); statement.setBytes(2,real2db(real)); try(final ResultSet rc=executeResultSet(statement)) { diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/EntryContainer.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/EntryContainer.java index c258484086..8909e56b69 100644 --- a/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/EntryContainer.java +++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/EntryContainer.java @@ -239,15 +239,23 @@ public ConfigChangeResult applyConfigurationDelete(final BackendIndexCfg cfg) EntryContainer.this.lock(); try { - storage.write(new WriteOperation() + // The write may be replayed by the storage, so the maps are updated outside of it: left inside, the second + // attempt would find nothing to delete and commit an empty transaction, reporting success for work that + // did not happen. The index may already be gone, since applyConfigurationAdd can fail after the config + // entry was persisted but before the index reached the map. + final AttributeIndex index = attrIndexMap.remove(cfg.getAttribute()); + attrCryptoMap.remove(cfg.getAttribute()); + if (index != null) { - @Override - public void run(WriteableTransaction txn) throws Exception + storage.write(new WriteOperation() { - attrIndexMap.remove(cfg.getAttribute()).closeAndDelete(txn); - attrCryptoMap.remove(cfg.getAttribute()); - } - }); + @Override + public void run(WriteableTransaction txn) throws Exception + { + index.closeAndDelete(txn); + } + }); + } } catch (Exception de) { @@ -321,14 +329,20 @@ public ConfigChangeResult applyConfigurationDelete(final BackendVLVIndexCfg cfg) EntryContainer.this.lock(); try { - storage.write(new WriteOperation() + // Removed outside the write for the reason given in the index delete listener above: the write may be + // replayed, and a replay must still have the deletion to perform. + final VLVIndex vlvIndex = vlvIndexMap.remove(cfg.getName().toLowerCase()); + if (vlvIndex != null) { - @Override - public void run(WriteableTransaction txn) throws Exception + storage.write(new WriteOperation() { - vlvIndexMap.remove(cfg.getName().toLowerCase()).closeAndDelete(txn); - } - }); + @Override + public void run(WriteableTransaction txn) throws Exception + { + vlvIndex.closeAndDelete(txn); + } + }); + } } catch (Exception e) { diff --git a/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStorageRetryTest.java b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStorageRetryTest.java new file mode 100644 index 0000000000..b2d9cb4d5d --- /dev/null +++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStorageRetryTest.java @@ -0,0 +1,131 @@ +/* + * 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.opends.server.backends.pluggable.spi.StorageRuntimeException; +import org.opends.server.types.DirectoryException; +import org.testng.annotations.DataProvider; +import org.testng.annotations.Test; + +import java.sql.SQLException; + +import static org.forgerock.i18n.LocalizableMessage.raw; +import static org.forgerock.opendj.ldap.ResultCode.OTHER; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + +/** + * Tests how a failure is classified as a transaction conflict, which is what decides whether + * {@link JDBCStorage#write} replays the operation, and how long it waits before it does. + *

+ * Runs without a database: the failures the drivers report are reproduced as synthetic + * {@link SQLException}s carrying the same vendor error number and SQLState. + */ +@Test(sequential = true) +@SuppressWarnings("javadoc") +public class JDBCStorageRetryTest extends DirectoryServerTestCase +{ + /** Driver class names, which is what the classification keys the vendor error numbers off. */ + private static final String MSSQL = "com.microsoft.sqlserver.jdbc.SQLServerConnection"; + private static final String MYSQL = "com.mysql.cj.jdbc.ConnectionImpl"; + private static final String ORACLE = "oracle.jdbc.driver.T4CConnection"; + private static final String POSTGRES = "org.postgresql.jdbc.PgConnection"; + + /** A failure whose cause chain is a cycle, to check that walking it terminates. */ + private static final class SelfCausedException extends RuntimeException + { + private static final long serialVersionUID = 1L; + + @Override + public synchronized Throwable getCause() + { + return this; + } + } + + @DataProvider + public Object[][] failures() + { + return new Object[][] { + // SQL Server picking a transaction as the deadlock victim: the failure this retry exists for + { "mssql deadlock victim", sql(1205, "40001"), MSSQL, true }, + // a deployment may add xopenStates=true to its connection URL, which reports the same deadlock as 42000 + { "mssql deadlock victim, xopenStates", sql(1205, "42000"), MSSQL, true }, + // the conflict of most other engines is carried by the SQLState, under a vendor number of their own + { "postgres serialization failure", sql(0, "40001"), POSTGRES, true }, + { "postgres deadlock detected", sql(0, "40P01"), POSTGRES, true }, + { "mysql deadlock", sql(1213, "40001"), MYSQL, true }, + // not a deadlock, but transient in the same way and equally resolved by a replay + { "mysql lock wait timeout", sql(1205, "HY000"), MYSQL, true }, + // Oracle maps ORA-00060 to SQLState 61000, so only its error number identifies the deadlock + { "oracle deadlock detected", sql(60, "61000"), ORACLE, true }, + + // the conflict reaches JDBCStorage.write() wrapped, so the whole cause chain has to be walked + { "wrapped once", new StorageRuntimeException(sql(1205, "40001")), MSSQL, true }, + { "wrapped twice", + new DirectoryException(OTHER, raw("unchecked"), new StorageRuntimeException(sql(1205, "40001"))), MSSQL, + true }, + + // the vendor numbers collide across engines, so they must not be matched driver-independently: + // ORA-01205 "not a data file" is fatal, and no replay resolves it + { "oracle not a data file", sql(1205, "64000"), ORACLE, false }, + // and a lock wait timeout is a MySQL number: 1205 means nothing of the kind to PostgreSQL + { "postgres unrelated 1205", sql(1205, "22001"), POSTGRES, false }, + + // nothing a replay can resolve + { "primary key violation", sql(2627, "23000"), MSSQL, false }, + { "syntax error", sql(102, "S0001"), MSSQL, false }, + { "no SQLState", sql(0, null), MSSQL, false }, + { "not a SQLException", new IllegalStateException("connection closed"), MSSQL, false }, + { "wrapped, not a conflict", new StorageRuntimeException(sql(2627, "23000")), MSSQL, false }, + { "no failure at all", null, MSSQL, false }, + { "unknown driver", sql(1205, "HY000"), null, false }, + { "cyclic cause chain", new SelfCausedException(), MSSQL, false }, + }; + } + + @Test(dataProvider = "failures") + public void testIsRetryableConflict(String name, Throwable failure, String driver, boolean expected) + { + assertEquals(JDBCStorage.isRetryableConflict(failure, driver), expected, name); + } + + /** The delay grows with the attempt, so that the replays outlast a contention lasting more than a few ms. */ + @Test + public void testRetryDelayGrowsAndStaysBounded() + { + long previousBound = 0; + for (int attempt = 1; attempt <= 10; attempt++) + { + long bound = 0; + for (int i = 0; i < 100; i++) + { + final long delay = JDBCStorage.retryDelayMillis(attempt); + assertTrue(delay >= 0, "attempt " + attempt + " waited " + delay + " ms"); + assertTrue(delay < 1000, "attempt " + attempt + " waited " + delay + " ms"); + bound = Math.max(bound, delay); + } + assertTrue(bound >= previousBound / 2, "attempt " + attempt + " did not grow past attempt " + (attempt - 1)); + previousBound = bound; + } + } + + private static SQLException sql(int errorCode, String sqlState) + { + return new SQLException("synthetic failure", sqlState, errorCode); + } +}