From 0b9c0f63f5c79e0a5d955011453bf415cb27e184 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Thu, 20 Aug 2026 09:17:16 +0000
Subject: [PATCH] Seek the primary key in the SQL Server upsert and retry a transaction conflict (#867)
---
opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/EntryContainer.java | 83 ++++++--
opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/TestCase.java | 80 ++++++++
opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStorageRetryTest.java | 165 ++++++++++++++++
opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java | 239 +++++++++++++++++++++--
4 files changed, 520 insertions(+), 47 deletions(-)
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 f0e6557..d1d8055 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
@@ -49,6 +49,47 @@
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;
+
+ /**
+ * Wall-clock budget the replays of a {@link #write} may spend, in nanoseconds. It is checked between attempts,
+ * so an attempt already running is never interrupted: the loop returns after at most this window plus one
+ * attempt. It bounds the conflicts that are slow to report, which {@link #MAX_RETRIES} alone does not - MySQL
+ * reports a lock wait timeout only after innodb_lock_wait_timeout, 50 s by default and not overridden here, so
+ * ten attempts would park a worker thread for eight minutes where a single one released it after 50 s. The
+ * deadlocks this retry exists for keep their full attempt budget, since every engine reports one in well under
+ * a second.
+ */
+ private static final long MAX_RETRY_WINDOW_NANOS = 10L * 1000L * 1000L * 1000L; //10 s
+
+ /** 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;
+
+ /** 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;
+
+ /**
+ * Class 40 states that are transaction rollbacks but must not be replayed. 40003 leaves the outcome of the
+ * transaction unknown, so replaying an add that in fact committed would answer the client with
+ * "entry already exists", and 40002 is an integrity constraint violation, which a replay repeats rather than
+ * resolves. Neither is reachable with the drivers shipped here - of class 40, Connector/J emits only 40000 and
+ * 40001, Oracle only ORA-02091/02092, and mssql-jdbc and PostgreSQL report their deadlock as 40001 and 40P01 -
+ * so they are excluded from the blanket class 40 match rather than that match being narrowed to a whitelist,
+ * which would fail a further engine reporting a conflict of its own.
+ */
+ private static final Set<String> NON_REPLAYABLE_ROLLBACK_STATES =
+ Collections.unmodifiableSet(new HashSet<>(Arrays.asList("40002", "40003")));
+
private JDBCBackendCfg config;
public JDBCStorage(JDBCBackendCfg cfg, ServerContext serverContext) {
@@ -251,8 +292,10 @@
}
}
+ /** Returns the class name of the driver behind the given connection, which names the engine it talks to. */
static String driverNameOf(Connection con) {
- return ((CachedConnection) con).parent.getClass().getName();
+ // a stamp connection comes straight from the driver, a transaction one from the pool
+ return ((con instanceof CachedConnection) ? ((CachedConnection) con).parent : con).getClass().getName();
}
// The dialect behind a pooled connection, or null for an engine none of the statements of this
@@ -767,6 +810,18 @@
}
//operation
+ /**
+ * {@inheritDoc}
+ * <p>
+ * A rolled back read is <em>not</em> replayed, as
+ * {@link org.opends.server.backends.pluggable.spi.Storage#read(ReadOperation)} requires: 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> T read(ReadOperation<T> readOperation) throws Exception {
try(final Connection con=getConnection()) {
@@ -774,24 +829,142 @@
}
}
+ /**
+ * {@inheritDoc}
+ * <p>
+ * {@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. It is bounded twice - by {@link #MAX_RETRIES} attempts and by the
+ * {@link #MAX_RETRY_WINDOW_NANOS} wall-clock window - because an attempt is not guaranteed to be short: a
+ * conflict an engine reports only after its own lock wait timeout would otherwise multiply that wait by the
+ * attempt count. A conflict that slow consumes the whole window in one attempt and is not replayed, which is
+ * what master did with it.
+ * <p>
+ * 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()) {
- final WriteableTransactionTransactionImpl txn=new WriteableTransactionTransactionImpl(con);
- try {
- writeOperation.run(txn);
- con.commit();
- } catch (Exception e) {
+ final long giveUpAt=System.nanoTime()+MAX_RETRY_WINDOW_NANOS;
+ for (int attempt=1;;attempt++) {
+ Exception failure=null;
+ String driver=null;
+ try (final Connection con=getConnection()) {
+ driver=driverNameOf(con);
+ final WriteableTransactionTransactionImpl txn=new WriteableTransactionTransactionImpl(con);
try {
- con.rollback();
- } catch (SQLException ex) {}
- throw e;
- } finally { // the comment connection lives no longer than the trees it stamped
- txn.stampSession.close();
+ writeOperation.run(txn);
+ 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;
+ } finally { // the comment connection lives no longer than the trees it stamped, and no longer
+ // than the attempt that opened it: a replay stamps on a session of its own
+ txn.stampSession.close();
+ }
+ } 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;
+ }
+ }
+ //System.nanoTime()-giveUpAt is the overflow safe form of the comparison
+ if (attempt>=MAX_RETRIES || System.nanoTime()-giveUpAt>=0 || !isRetryableConflict(failure,driver)) {
+ throw failure;
+ }
+ //logged rather than silently absorbed, so that a deployment retrying most of its writes stays observable;
+ //one line per replay, since an add can emit nine of them and a stack trace each time reads as a failure
+ logger.warn(LocalizableMessage.raw("jdbc: replaying the transaction after a conflict, attempt %d of %d: %s",
+ attempt, MAX_RETRIES, conflictSummary(failure)));
+ if (logger.isTraceEnabled()) {
+ logger.trace("jdbc: the conflict being replayed was %s", 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 whether the given failure carries a transaction conflict that replaying the operation can resolve.
+ * <p>
+ * 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.
+ * <p>
+ * The standard class 40 states carry the conflict of most engines - 40P01 for PostgreSQL, 40001 for SQL Server
+ * and for MySQL, whose driver replaces the server side HY000 of a deadlock and of a lock wait timeout with
+ * 40001 - 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. 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. MySQL needs no number of its own, since its driver has already
+ * mapped both conditions into class 40; see {@link #NON_REPLAYABLE_ROLLBACK_STATES} for the two class 40 states
+ * that are excluded from that match.
+ */
+ static boolean isRetryableConflict(Throwable t, String driver) {
+ for (int hop=0; t!=null && hop<MAX_CAUSE_HOPS; t=t.getCause(), hop++) {
+ if (t instanceof SQLException && isConflict((SQLException) t, driver)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean isConflict(SQLException e, String driver) {
+ final String state=String.valueOf(e.getSQLState());
+ if (state.startsWith("40") && !NON_REPLAYABLE_ROLLBACK_STATES.contains(state)) {
+ return true;
+ }
+ final String driverName=String.valueOf(driver);
+ if (driverName.contains("oracle")) {
+ return e.getErrorCode()==ORACLE_DEADLOCK_DETECTED;
+ } else if (driverName.contains("microsoft")) {
+ return e.getErrorCode()==MSSQL_DEADLOCK_VICTIM;
+ }
+ return false;
+ }
+
+ /**
+ * Returns the SQLState and vendor error number of the first {@link SQLException} of the given cause chain, which
+ * is what identifies a conflict, so that a replay can be logged without a stack trace on every attempt.
+ */
+ static String conflictSummary(Throwable failure) {
+ Throwable t=failure;
+ for (int hop=0; t!=null && hop<MAX_CAUSE_HOPS; t=t.getCause(), hop++) {
+ if (t instanceof SQLException) {
+ final SQLException e=(SQLException) t;
+ return "SQLState "+e.getSQLState()+", error "+e.getErrorCode()+": "+e.getMessage();
+ }
+ }
+ return String.valueOf(failure);
+ }
+
static final byte[] NULL=new byte[]{(byte)0};
static byte[] real2db(byte[] real) {
@@ -819,6 +992,21 @@
}
});
+ /**
+ * Returns the placeholder to compare against the {@code h} column, casting it where the driver would
+ * otherwise bind a value of the wrong type.
+ * <p>
+ * 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 driverNameOf(con).contains("microsoft") ? "cast(? as char(128))" : "?";
+ }
+
private class ReadableTransactionImpl implements ReadableTransaction {
final Connection con;
boolean isReadOnly=true;
@@ -829,7 +1017,7 @@
@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)) {
@@ -895,11 +1083,11 @@
}
String getTableDialect() {
- if (((CachedConnection) con).parent.getClass().getName().contains("oracle")) {
+ if (driverNameOf(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 (driverNameOf(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 (driverNameOf(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)";
@@ -917,7 +1105,7 @@
}
}
// 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=driverNameOf(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)")){
@@ -998,12 +1186,15 @@
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=driverNameOf(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())));
@@ -1025,8 +1216,8 @@
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());
@@ -1075,7 +1266,7 @@
@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);
@@ -1194,7 +1385,7 @@
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);
@@ -1238,7 +1429,7 @@
@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 c258484..e9188ab 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
@@ -38,6 +38,8 @@
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -199,21 +201,28 @@
{
final CryptoSuite cryptoSuite = newCryptoSuite(cfg.isConfidentialityEnabled());
final AttributeIndex index = newAttributeIndex(cfg, cryptoSuite);
+ final AtomicBoolean trusted = new AtomicBoolean();
storage.write(new WriteOperation()
{
@Override
public void run(WriteableTransaction txn) throws Exception
{
+ // The write may be replayed by the storage, and open() registers this index as a change listener of
+ // its configuration. close() removes every registration made for this index, so closing first leaves
+ // one listener behind rather than one per attempt; it is a no-op on the first attempt.
+ index.close();
index.open(txn, true);
- if (!index.isTrusted())
- {
- ccr.setAdminActionRequired(true);
- ccr.addMessage(NOTE_INDEX_ADD_REQUIRES_REBUILD.get(cfg.getAttribute().getNameOrOID()));
- }
+ trusted.set(index.isTrusted());
attrIndexMap.put(cfg.getAttribute(), index);
attrCryptoMap.put(cfg.getAttribute(), cryptoSuite);
}
});
+ if (!trusted.get())
+ {
+ // Reported outside the write, since a replayed attempt would otherwise repeat the message.
+ ccr.setAdminActionRequired(true);
+ ccr.addMessage(NOTE_INDEX_ADD_REQUIRES_REBUILD.get(cfg.getAttribute().getNameOrOID()));
+ }
}
catch(Exception e)
{
@@ -239,15 +248,23 @@
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)
{
@@ -283,21 +300,35 @@
final ConfigChangeResult ccr = new ConfigChangeResult();
try
{
+ final AtomicReference<VLVIndex> built = new AtomicReference<>();
+ final AtomicBoolean trusted = new AtomicBoolean();
storage.write(new WriteOperation()
{
@Override
public void run(WriteableTransaction txn) throws Exception
{
- VLVIndex vlvIndex = new VLVIndex(cfg, state, storage, EntryContainer.this, txn);
- vlvIndex.open(txn, true);
- if(!vlvIndex.isTrusted())
+ // The write may be replayed by the storage, and the VLVIndex constructor registers the new instance as
+ // a change listener of its configuration. Only the last instance reaches the map, so the one built by
+ // the previous attempt is closed here, which deregisters it: left registered it would never be closed
+ // again, and every later VLV configuration change would be applied once per attempt.
+ final VLVIndex previous = built.getAndSet(null);
+ if (previous != null)
{
- ccr.setAdminActionRequired(true);
- ccr.addMessage(NOTE_INDEX_ADD_REQUIRES_REBUILD.get(cfg.getName()));
+ previous.close();
}
+ VLVIndex vlvIndex = new VLVIndex(cfg, state, storage, EntryContainer.this, txn);
+ built.set(vlvIndex);
+ vlvIndex.open(txn, true);
+ trusted.set(vlvIndex.isTrusted());
vlvIndexMap.put(cfg.getName().toLowerCase(), vlvIndex);
}
});
+ if (!trusted.get())
+ {
+ // Reported outside the write, since a replayed attempt would otherwise repeat the message.
+ ccr.setAdminActionRequired(true);
+ ccr.addMessage(NOTE_INDEX_ADD_REQUIRES_REBUILD.get(cfg.getName()));
+ }
}
catch(Exception e)
{
@@ -321,14 +352,20 @@
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 0000000..eac8048
--- /dev/null
+++ b/opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStorageRetryTest.java
@@ -0,0 +1,165 @@
+/*
+ * 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.
+ * <p>
+ * 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 },
+ // Connector/J replaces the server side HY000 of both conditions with 40001, so neither needs a number here
+ { "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, "40001"), MYSQL, true },
+ // the rollback a MySQL group replication conflict reports, error 3101, which the driver maps to 40000
+ { "mysql group replication rollback", sql(3101, "40000"), 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 },
+
+ // two class 40 states are rollbacks that a replay must not repeat: 40003 leaves the outcome of the
+ // transaction unknown, and 40002 is an integrity constraint violation that a replay would only hit again
+ { "statement completion unknown", sql(0, "40003"), POSTGRES, false },
+ { "transaction integrity constraint violation", sql(0, "40002"), POSTGRES, false },
+ // ... but the state of a conflict is still matched whatever vendor number carries it
+ { "class 40 is driver independent", sql(0, "40001"), null, true },
+
+ // 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 },
+ // a vendor number is never matched without a driver to key it off, since the engines collide on it
+ { "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;
+ }
+ }
+
+ /**
+ * A replay is logged once per attempt, so what it logs has to identify the conflict without a stack trace: the
+ * SQLState and the vendor error number, reached through however many wrappers the failure arrived in.
+ */
+ @Test
+ public void testConflictSummaryNamesTheStateAndTheNumber()
+ {
+ final String summary = JDBCStorage.conflictSummary(
+ new DirectoryException(OTHER, raw("unchecked"), new StorageRuntimeException(sql(1205, "40001"))));
+ assertTrue(summary.contains("40001"), summary);
+ assertTrue(summary.contains("1205"), summary);
+ assertTrue(summary.contains("synthetic failure"), summary);
+ }
+
+ /** A failure carrying no SQLException at all, and a cyclic cause chain, still have to yield something loggable. */
+ @Test
+ public void testConflictSummaryTerminatesWithoutASQLException()
+ {
+ assertTrue(JDBCStorage.conflictSummary(new IllegalStateException("connection closed")).contains("closed"));
+ assertTrue(JDBCStorage.conflictSummary(new SelfCausedException()).contains("SelfCausedException"));
+ assertEquals(JDBCStorage.conflictSummary(null), "null");
+ }
+
+ private static SQLException sql(int errorCode, String sqlState)
+ {
+ return new SQLException("synthetic failure", sqlState, errorCode);
+ }
+}
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 76760c1..b650301 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
@@ -45,6 +45,10 @@
import java.util.Collections;
import java.util.List;
import java.util.NoSuchElementException;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import static org.forgerock.opendj.config.ConfigurationMock.mockCfg;
@@ -1236,4 +1240,80 @@
storage.close();
}
}
+ /**
+ * Two or more <em>distinct new</em> keys written into one tree per transaction is the shape the primary key
+ * seek made able to deadlock: on the NOT MATCHED path the seek range-locks the gap before the next existing
+ * key, that lock is self-incompatible, and the key hash scatters logically ordered keys across the index, so
+ * two writers inserting different keys can each end up holding what the other needs. The ascending key order
+ * that {@code IndexBuffer} maintains does not help there. Nothing may escape {@link JDBCStorage#write}, which
+ * replays the conflict, and no record may be lost to it (#867).
+ */
+ @Test(timeOut = 600000)
+ public void testConcurrentWritersInsertingDistinctKeys() throws Exception {
+ final int writers = 4;
+ final int rounds = 25;
+ final int keysPerTransaction = 3;
+ final int seeded = 10;
+ final JDBCStorage storage = new JDBCStorage(createBackendCfg(), null);
+ final TreeName tree = new TreeName("testConcurrentInsert", "tree");
+ final ExecutorService executor = Executors.newFixedThreadPool(writers);
+ try {
+ storage.open(AccessMode.READ_WRITE);
+ // seeded, so that every insert below takes the NOT MATCHED path with a gap to lock in front of it
+ storage.write(new WriteOperation() {
+ @Override
+ public void run(WriteableTransaction txn) throws Exception {
+ txn.openTree(tree, true);
+ for (int i = 0; i < seeded; i++) {
+ txn.put(tree, key(i), value(i));
+ }
+ }
+ });
+ final List<Callable<Void>> concurrent = new ArrayList<>();
+ for (int writer = 0; writer < writers; writer++) {
+ final int id = writer;
+ concurrent.add(new Callable<Void>() {
+ @Override
+ public Void call() throws Exception {
+ for (int round = 0; round < rounds; round++) {
+ final int current = round;
+ storage.write(new WriteOperation() {
+ @Override
+ public void run(WriteableTransaction txn) throws Exception {
+ for (int i = 0; i < keysPerTransaction; i++) {
+ txn.put(tree,
+ ByteString.valueOfUtf8(String.format("w%02d-r%03d-k%d", id, current, i)),
+ value(i));
+ }
+ }
+ });
+ }
+ return null;
+ }
+ });
+ }
+ for (final Future<Void> written : executor.invokeAll(concurrent)) {
+ // a conflict the storage did not replay surfaces here, as it would reach an LDAP client
+ written.get();
+ }
+ storage.read(new ReadOperation<Void>() {
+ @Override
+ public Void run(ReadableTransaction txn) throws Exception {
+ assertEquals(txn.getRecordCount(tree), seeded + writers * rounds * keysPerTransaction);
+ return null;
+ }
+ });
+ } finally {
+ executor.shutdownNow();
+ try {
+ storage.write(new WriteOperation() {
+ @Override
+ public void run(WriteableTransaction txn) throws Exception {
+ txn.deleteTree(tree);
+ }
+ });
+ } catch (Exception ignored) {}
+ storage.close();
+ }
+ }
}
--
Gitblit v1.10.0