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

opendj-server-legacy/src/main/java/org/opends/server/backends/jdbc/JDBCStorage.java

@@ -47,6 +47,18 @@ public class JDBCStorage implements org.opends.server.backends.pluggable.spi.Sto
 
 	private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
 
+	/** Number of attempts a {@link #read} or {@link #write} makes before it propagates the conflict to the caller. */
+	private static final int MAX_RETRIES = 10;
+
+	/** Upper bound of the random delay inserted between two attempts, in milliseconds. */
+	private static final double MAX_SLEEP_ON_RETRY_MS = 50.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 ERROR_DEADLOCK_VICTIM = 1205;
+
 	private JDBCBackendCfg config;
 
 	public JDBCStorage(JDBCBackendCfg cfg, ServerContext serverContext) {
@@ -172,26 +184,86 @@ public void removeStorageFiles() throws StorageRuntimeException {
 	//operation
 	@Override
 	public <T> T read(ReadOperation<T> readOperation) throws Exception {
-		try(final Connection con=getConnection()) {
-			return readOperation.run(new ReadableTransactionImpl(con));
+		for (int attempt=1;;attempt++) {
+			try(final Connection con=getConnection()) {
+				return readOperation.run(new ReadableTransactionImpl(con));
+			} catch (Exception e) {
+				if (!retryOnConflict(e,attempt)) {
+					throw e;
+				}
+			}
 		}
 	}
 
 	@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++) {
+			try (final Connection con=getConnection()) {
 				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) {}
+					throw e;
+				}
+			} catch (Exception e) {
+				if (!retryOnConflict(e,attempt)) {
+					throw e;
+				}
 			}
 		}
 	}
 
+	/**
+	 * Waits for a short random delay and returns whether the failed attempt should be replayed.
+	 * <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. The delay is randomized to spread the retries of the transactions that collided.
+	 */
+	private boolean retryOnConflict(Exception e, int attempt) throws InterruptedException {
+		if (attempt>=MAX_RETRIES || !isRetryableConflict(e)) {
+			return false;
+		}
+		//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(e)));
+		Thread.sleep((long) (Math.random() * MAX_SLEEP_ON_RETRY_MS));
+		return true;
+	}
+
+	/**
+	 * 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>
+	 * Both the vendor error number and the SQLState are examined, since neither alone covers the drivers in use.
+	 * SQL Server reports the deadlock victim as error 1205 with SQLState 40001, but as 42000 when the connection was
+	 * opened with {@code xopenStates=true}, so the state cannot be relied upon; conversely the standard class 40
+	 * states carry the conflict of the other engines - 40P01 for PostgreSQL, 40001 for MySQL and H2 - under vendor
+	 * error numbers of their own. Error 1205 of MySQL is a lock wait timeout rather than a deadlock, which the code
+	 * alone cannot tell apart, but that condition is transient as well and is equally resolved by a replay.
+	 */
+	static boolean isRetryableConflict(Throwable t) {
+		for (int hop=0; t!=null && hop<MAX_CAUSE_HOPS; t=t.getCause(), hop++) {
+			if (t instanceof SQLException) {
+				final SQLException e=(SQLException) t;
+				if (e.getErrorCode()==ERROR_DEADLOCK_VICTIM || String.valueOf(e.getSQLState()).startsWith("40")) {
+					return true;
+				}
+			}
+		}
+		return false;
+	}
+
 	static final byte[] NULL=new byte[]{(byte)0};
 
 	static byte[] real2db(byte[] real) {
@@ -219,6 +291,21 @@ static byte[] db2real(byte[] db) {
 			}
 		});
 
+	/**
+	 * 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 ((CachedConnection) con).parent.getClass().getName().contains("microsoft") ? "cast(? as char(128))" : "?";
+	}
+
 	private class ReadableTransactionImpl implements ReadableTransaction {
 		final Connection con;
 		boolean isReadOnly=true;
@@ -229,7 +316,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)) {
@@ -376,7 +463,10 @@ 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);
 			}
 		}
 
@@ -403,8 +493,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 +543,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 +662,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 +706,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)) {

opendj-server-legacy/src/main/java/org/opends/server/backends/pluggable/EntryContainer.java

@@ -244,7 +244,12 @@ public ConfigChangeResult applyConfigurationDelete(final BackendIndexCfg cfg)
           @Override
           public void run(WriteableTransaction txn) throws Exception
           {
-            attrIndexMap.remove(cfg.getAttribute()).closeAndDelete(txn);
+            // The write may be replayed by the storage, so the removal must tolerate having already happened.
+            final AttributeIndex index = attrIndexMap.remove(cfg.getAttribute());
+            if (index != null)
+            {
+              index.closeAndDelete(txn);
+            }
             attrCryptoMap.remove(cfg.getAttribute());
           }
         });
@@ -326,7 +331,12 @@ public ConfigChangeResult applyConfigurationDelete(final BackendVLVIndexCfg cfg)
           @Override
           public void run(WriteableTransaction txn) throws Exception
           {
-            vlvIndexMap.remove(cfg.getName().toLowerCase()).closeAndDelete(txn);
+            // The write may be replayed by the storage, so the removal must tolerate having already happened.
+            final VLVIndex vlvIndex = vlvIndexMap.remove(cfg.getName().toLowerCase());
+            if (vlvIndex != null)
+            {
+              vlvIndex.closeAndDelete(txn);
+            }
           }
         });
       }

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

@@ -0,0 +1,94 @@
+/*
+ * 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;
+
+/**
+ * Tests how a failure is classified as a transaction conflict, which is what decides whether
+ * {@link JDBCStorage#write} and {@link JDBCStorage#read} replay the operation.
+ * <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
+{
+  /** 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"), true },
+      // the xopenStates connection property makes the same driver report a state of no help here
+      { "mssql deadlock victim, xopenStates", sql(1205, "42000"), true },
+      // the conflict of the other engines is carried by the SQLState, under a vendor number of their own
+      { "postgres serialization failure", sql(0, "40001"), true },
+      { "postgres deadlock detected", sql(0, "40P01"), true },
+      { "mysql deadlock", sql(1213, "40001"), true },
+      // not a deadlock, but transient in the same way and equally resolved by a replay
+      { "mysql lock wait timeout", sql(1205, "HY000"), true },
+
+      // the conflict reaches JDBCStorage.write() wrapped, so the whole cause chain has to be walked
+      { "wrapped once", new StorageRuntimeException(sql(1205, "40001")), true },
+      { "wrapped twice",
+        new DirectoryException(OTHER, raw("unchecked"), new StorageRuntimeException(sql(1205, "40001"))), true },
+
+      // nothing a replay can resolve
+      { "primary key violation", sql(2627, "23000"), false },
+      { "syntax error", sql(102, "S0001"), false },
+      { "no SQLState", sql(0, null), false },
+      { "not a SQLException", new IllegalStateException("connection closed"), false },
+      { "wrapped, not a conflict", new StorageRuntimeException(sql(2627, "23000")), false },
+      { "no failure at all", null, false },
+      { "cyclic cause chain", new SelfCausedException(), false },
+    };
+  }
+
+  @Test(dataProvider = "failures")
+  public void testIsRetryableConflict(String name, Throwable failure, boolean expected)
+  {
+    assertEquals(JDBCStorage.isRetryableConflict(failure), expected, name);
+  }
+
+  private static SQLException sql(int errorCode, String sqlState)
+  {
+    return new SQLException("synthetic failure", sqlState, errorCode);
+  }
+}