Seek the primary key in the SQL Server upsert and retry a transaction confl...

Valery Kharseko

14 hours ago 0b9c0f63f5c79e0a5d955011453bf415cb27e184

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/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)) {

 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)
      {

 opendj-server-legacy/src/test/java/org/opends/server/backends/jdbc/JDBCStorageRetryTest.java

New file
@@ -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);
  }
}

 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();
        }
    }
}