Add the ability to specify per-transaction isolation level (#2104)

A config file field is added to control if per-transaction isolation
level is actually used. If set to true, nested transactions will throw
a runtime exception as the enclosing transaction may run at a different
isolation level.

In this PR we only add the ability to specify the isolation level,
without enabling it in any environment (including unit tests), or
actually specifying one for any query. This should allow us to set up
the system without impacting anything currently working.

core/src/main/java/google/registry/config/RegistryConfig.java

@@ -1547,6 +1547,11 @@ public final class RegistryConfig {
     return CONFIG_SETTINGS.get().hibernate.connectionIsolation;
   }
 
+  /** Returns true if per-transaction isolation level is enabled. */
+  public static boolean getHibernatePerTransactionIsolationEnabled() {
+    return CONFIG_SETTINGS.get().hibernate.perTransactionIsolation;
+  }
+
   /** Returns true if hibernate.show_sql is enabled. */
   public static String getHibernateLogSqlQueries() {
     return CONFIG_SETTINGS.get().hibernate.logSqlQueries;

core/src/main/java/google/registry/config/RegistryConfigSettings.java

@@ -115,6 +115,7 @@ public class RegistryConfigSettings {
 
   /** Configuration for Hibernate. */
   public static class Hibernate {
+    public boolean perTransactionIsolation;
     public String connectionIsolation;
     public String logSqlQueries;
     public String hikariConnectionTimeout;

core/src/main/java/google/registry/config/files/default-config.yaml

@@ -189,6 +189,12 @@ registryPolicy:
   sunriseDomainCreateDiscount: 0.15
 
 hibernate:
+  # Make it possible to specify the isolation level for each transaction. If set
+  # to true, nested transactions will throw an exception. If set to false, a
+  # transaction with the isolation override specified will still execute at the
+  # default level (specified below).
+  perTransactionIsolation: false
+
   # Make 'SERIALIZABLE' the default isolation level to ensure correctness.
   #
   # Entities that are never involved in multi-table transactions may use optimistic
@@ -479,7 +485,7 @@ keyring:
 
 # Configuration options relevant to the "nomulus" registry tool.
 registryTool:
-  # OAuth client Id used by the tool.
+  # OAuth client ID used by the tool.
   clientId: YOUR_CLIENT_ID
   # OAuth client secret used by the tool.
   clientSecret: YOUR_CLIENT_SECRET

core/src/main/java/google/registry/config/files/nomulus-config-unittest.yaml

@@ -26,3 +26,6 @@ gSuite:
 misc:
   # We would rather have failures than timeouts, so reduce the number of retries
   transientFailureRetries: 3
+
+hibernate:
+  perTransactionIsolation: false

core/src/main/java/google/registry/flows/FlowModule.java

@@ -35,6 +35,8 @@ import google.registry.model.eppoutput.EppResponse;
 import google.registry.model.eppoutput.Result;
 import google.registry.model.host.HostHistory;
 import google.registry.model.reporting.HistoryEntry;
+import google.registry.persistence.IsolationLevel;
+import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
 import java.lang.annotation.Documented;
 import java.util.Optional;
 import javax.inject.Qualifier;
@@ -135,6 +137,14 @@ public class FlowModule {
     return TransactionalFlow.class.isAssignableFrom(flowClass);
   }
 
+  @Provides
+  @FlowScope
+  Optional<TransactionIsolationLevel> provideIsolationLevelOverride(
+      Class<? extends Flow> flowClass) {
+    return Optional.ofNullable(flowClass.getAnnotation(IsolationLevel.class))
+        .map(IsolationLevel::value);
+  }
+
   @Provides
   @FlowScope
   @Superuser
@@ -166,7 +176,7 @@ public class FlowModule {
   @FlowScope
   @RegistrarId
   static String provideRegistrarId(SessionMetadata sessionMetadata) {
-    // Treat a missing registrarId as null so we can always inject a non-null value. All we do with
+    // Treat a missing registrarId as null, so we can always inject a non-null value. All we do with
     // the registrarId is log it (as "") or detect its absence, both of which work fine with empty.
     return Strings.nullToEmpty(sessionMetadata.getRegistrarId());
   }

core/src/main/java/google/registry/flows/FlowRunner.java

@@ -28,6 +28,8 @@ import google.registry.flows.session.LoginFlow;
 import google.registry.model.eppcommon.Trid;
 import google.registry.model.eppoutput.EppOutput;
 import google.registry.monitoring.whitebox.EppMetric;
+import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
+import java.util.Optional;
 import javax.inject.Inject;
 import javax.inject.Provider;
 
@@ -42,6 +44,7 @@ public class FlowRunner {
   @Inject TransportCredentials credentials;
   @Inject EppRequestSource eppRequestSource;
   @Inject Provider<Flow> flowProvider;
+  @Inject Optional<TransactionIsolationLevel> isolationLevelOverride;
   @Inject Class<? extends Flow> flowClass;
   @Inject @InputXml byte[] inputXmlBytes;
   @Inject @DryRun boolean isDryRun;
@@ -91,7 +94,8 @@ public class FlowRunner {
                 } catch (EppException e) {
                   throw new EppRuntimeException(e);
                 }
-              });
+              },
+              isolationLevelOverride.orElse(null));
     } catch (DryRunException e) {
       return e.output;
     } catch (EppRuntimeException e) {

core/src/main/java/google/registry/persistence/IsolationLevel.java

@@ -0,0 +1,33 @@
+// Copyright 2023 The Nomulus Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package google.registry.persistence;
+
+import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Indicates which {@link TransactionIsolationLevel} that a {@link
+ * google.registry.flows.TransactionalFlow} show run at.
+ */
+@Documented
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+public @interface IsolationLevel {
+  TransactionIsolationLevel value();
+}

core/src/main/java/google/registry/persistence/PersistenceModule.java

@@ -27,9 +27,9 @@ import static google.registry.persistence.transaction.TransactionManagerFactory.
 
 import com.google.auth.oauth2.GoogleCredentials;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Ascii;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
-import com.google.common.flogger.FluentLogger;
 import dagger.BindsOptionalOf;
 import dagger.Module;
 import dagger.Provides;
@@ -65,7 +65,6 @@ import org.hibernate.cfg.Environment;
 /** Dagger module class for the persistence layer. */
 @Module
 public abstract class PersistenceModule {
-  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
   // This name must be the same as the one defined in persistence.xml.
   public static final String PERSISTENCE_UNIT_NAME = "nomulus";
@@ -166,12 +165,10 @@ public abstract class PersistenceModule {
     // Override the default minimum which is tuned for the Registry server. A worker VM should
     // release all connections if it no longer interacts with the database.
     overrides.put(HIKARI_MINIMUM_IDLE, "0");
-    /**
-     * Disable Hikari's maxPoolSize limit check by setting it to an absurdly large number. The
-     * effective (and desirable) limit is the number of pipeline threads on the pipeline worker,
-     * which can be configured using pipeline options. See {@link RegistryPipelineOptions} for more
-     * information.
-     */
+    // Disable Hikari's maxPoolSize limit check by setting it to an absurdly large number. The
+    // effective (and desirable) limit is the number of pipeline threads on the pipeline worker,
+    // which can be configured using pipeline options. See {@link RegistryPipelineOptions} for more
+    // information.
     overrides.put(HIKARI_MAXIMUM_POOL_SIZE, String.valueOf(Integer.MAX_VALUE));
     instanceConnectionNameOverride
         .map(Provider::get)
@@ -303,7 +300,7 @@ public abstract class PersistenceModule {
 
   private static EntityManagerFactory create(Map<String, String> properties) {
     // If there are no annotated classes, we can create the EntityManagerFactory from the generic
-    // method.  Otherwise we have to use a more tailored approach.  Note that this adds to the set
+    // method.  Otherwise, we have to use a more tailored approach.  Note that this adds to the set
     // of annotated classes defined in the configuration, it does not override them.
     EntityManagerFactory emf =
         Persistence.createEntityManagerFactory(
@@ -322,8 +319,7 @@ public abstract class PersistenceModule {
       overrides.put(Environment.USER, credential.login());
       overrides.put(Environment.PASS, credential.password());
     } catch (Throwable e) {
-      // TODO(b/184631990): after SQL becomes primary, throw an exception to fail fast
-      logger.atSevere().withCause(e).log("Failed to get SQL credential from Secret Manager.");
+      throw new RuntimeException("Failed to get SQL credential from Secret Manager.", e);
     }
   }
 
@@ -340,11 +336,13 @@ public abstract class PersistenceModule {
     TRANSACTION_SERIALIZABLE;
 
     private final int value;
+    private final String mode;
 
     TransactionIsolationLevel() {
       try {
         // name() is final in parent class (Enum.java), therefore safe to call in constructor.
         value = Connection.class.getField(name()).getInt(null);
+        mode = name().substring(12).replace('_', ' ');
       } catch (Exception e) {
         throw new IllegalStateException(
             String.format(
@@ -356,6 +354,14 @@ public abstract class PersistenceModule {
     public final int getValue() {
       return value;
     }
+
+    public final String getMode() {
+      return mode;
+    }
+
+    public static TransactionIsolationLevel fromMode(String mode) {
+      return valueOf(String.format("TRANSACTION_%s", Ascii.toUpperCase(mode.replace(' ', '_'))));
+    }
   }
 
   /** Types of {@link JpaTransactionManager JpaTransactionManagers}. */

core/src/main/java/google/registry/persistence/transaction/JpaTransactionManager.java

@@ -14,6 +14,7 @@
 
 package google.registry.persistence.transaction;
 
+import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
 import google.registry.persistence.VKey;
 import java.util.function.Supplier;
 import javax.persistence.EntityManager;
@@ -64,9 +65,21 @@ public interface JpaTransactionManager extends TransactionManager {
   /** Executes the work in a transaction with no retries and returns the result. */
   <T> T transactNoRetry(Supplier<T> work);
 
+  /**
+   * Executes the work in a transaction at the given {@link TransactionIsolationLevel} with no
+   * retries and returns the result.
+   */
+  <T> T transactNoRetry(Supplier<T> work, TransactionIsolationLevel isolationLevel);
+
   /** Executes the work in a transaction with no retries. */
   void transactNoRetry(Runnable work);
 
+  /**
+   * Executes the work in a transaction at the given {@link TransactionIsolationLevel} with no
+   * retries.
+   */
+  void transactNoRetry(Runnable work, TransactionIsolationLevel isolationLevel);
+
   /** Deletes the entity by its id, throws exception if the entity is not deleted. */
   <T> void assertDelete(VKey<T> key);
 
@@ -84,4 +97,13 @@ public interface JpaTransactionManager extends TransactionManager {
   static Query setQueryFetchSize(Query query, int fetchSize) {
     return query.setHint("org.hibernate.fetchSize", fetchSize);
   }
+
+  /** Return the default {@link TransactionIsolationLevel} specified via the config file. */
+  TransactionIsolationLevel getDefaultTransactionIsolationLevel();
+
+  /** Return the {@link TransactionIsolationLevel} used in the current transaction. */
+  TransactionIsolationLevel getCurrentTransactionIsolationLevel();
+
+  /** Asserts that the current transaction runs at the given level. */
+  void assertTransactionIsolationLevel(TransactionIsolationLevel expectedLevel);
 }

core/src/main/java/google/registry/persistence/transaction/JpaTransactionManagerImpl.java

@@ -18,6 +18,7 @@ import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
+import static google.registry.config.RegistryConfig.getHibernatePerTransactionIsolationEnabled;
 import static google.registry.util.PreconditionsUtils.checkArgumentNotNull;
 import static java.util.AbstractMap.SimpleEntry;
 import static java.util.stream.Collectors.joining;
@@ -31,6 +32,7 @@ import com.google.common.collect.Streams;
 import com.google.common.flogger.FluentLogger;
 import google.registry.model.ImmutableObject;
 import google.registry.persistence.JpaRetries;
+import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
 import google.registry.persistence.VKey;
 import google.registry.util.Clock;
 import google.registry.util.Retrier;
@@ -65,6 +67,7 @@ import javax.persistence.TemporalType;
 import javax.persistence.TypedQuery;
 import javax.persistence.criteria.CriteriaQuery;
 import javax.persistence.metamodel.EntityType;
+import org.hibernate.cfg.Environment;
 import org.joda.time.DateTime;
 
 /** Implementation of {@link JpaTransactionManager} for JPA compatible database. */
@@ -77,9 +80,6 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
   private final EntityManagerFactory emf;
   private final Clock clock;
 
-  // TODO(b/177588434): Investigate alternatives for managing transaction information. ThreadLocal
-  // adds an unnecessary restriction that each request has to be processed by one thread
-  // synchronously.
   private static final ThreadLocal<TransactionInfo> transactionInfo =
       ThreadLocal.withInitial(TransactionInfo::new);
 
@@ -137,18 +137,42 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
   }
 
   @Override
-  public <T> T transact(Supplier<T> work) {
-    // This prevents inner transaction from retrying, thus avoiding a cascade retry effect.
-    if (inTransaction()) {
-      return transactNoRetry(work);
+  public void assertTransactionIsolationLevel(TransactionIsolationLevel expectedLevel) {
+    assertInTransaction();
+    TransactionIsolationLevel currentLevel = getCurrentTransactionIsolationLevel();
+    if (currentLevel != expectedLevel) {
+      throw new IllegalStateException(
+          String.format(
+              "Current transaction isolation level (%s) is not as expected (%s)",
+              currentLevel, expectedLevel));
     }
-    return retrier.callWithRetry(() -> transactNoRetry(work), JpaRetries::isFailedTxnRetriable);
   }
 
   @Override
-  public <T> T transactNoRetry(Supplier<T> work) {
+  public <T> T transact(Supplier<T> work, TransactionIsolationLevel isolationLevel) {
+    // This prevents inner transaction from retrying, thus avoiding a cascade retry effect.
     if (inTransaction()) {
-      return work.get();
+      return transactNoRetry(work, isolationLevel);
+    }
+    return retrier.callWithRetry(
+        () -> transactNoRetry(work, isolationLevel), JpaRetries::isFailedTxnRetriable);
+  }
+
+  @Override
+  public <T> T transact(Supplier<T> work) {
+    return transact(work, null);
+  }
+
+  @Override
+  public <T> T transactNoRetry(
+      Supplier<T> work, @Nullable TransactionIsolationLevel isolationLevel) {
+    if (inTransaction()) {
+      if (getHibernatePerTransactionIsolationEnabled()) {
+        throw new IllegalStateException("Nested transaction detected");
+      } else {
+        logger.atWarning().log("Nested transaction detected");
+        return work.get();
+      }
     }
     TransactionInfo txnInfo = transactionInfo.get();
     txnInfo.entityManager = emf.createEntityManager();
@@ -156,6 +180,18 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
     try {
       txn.begin();
       txnInfo.start(clock);
+      if (isolationLevel != null) {
+        if (getHibernatePerTransactionIsolationEnabled()) {
+          getEntityManager()
+              .createNativeQuery(
+                  String.format("SET TRANSACTION ISOLATION LEVEL %s", isolationLevel.getMode()))
+              .executeUpdate();
+          logger.atInfo().log("Running transaction at %s", isolationLevel);
+        } else {
+          logger.atWarning().log(
+              "Per-transaction isolation level disabled, but %s was requested", isolationLevel);
+        }
+      }
       T result = work.get();
       txn.commit();
       return result;
@@ -174,21 +210,55 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
   }
 
   @Override
-  public void transact(Runnable work) {
+  public <T> T transactNoRetry(Supplier<T> work) {
+    return transactNoRetry(work, null);
+  }
+
+  @Override
+  public void transact(Runnable work, TransactionIsolationLevel isolationLevel) {
     transact(
         () -> {
           work.run();
           return null;
-        });
+        },
+        isolationLevel);
   }
 
   @Override
-  public void transactNoRetry(Runnable work) {
+  public void transact(Runnable work) {
+    transact(work, null);
+  }
+
+  @Override
+  public void transactNoRetry(Runnable work, TransactionIsolationLevel isolationLevel) {
     transactNoRetry(
         () -> {
           work.run();
           return null;
-        });
+        },
+        isolationLevel);
+  }
+
+  @Override
+  public void transactNoRetry(Runnable work) {
+    transactNoRetry(work, null);
+  }
+
+  @Override
+  public TransactionIsolationLevel getDefaultTransactionIsolationLevel() {
+    return TransactionIsolationLevel.valueOf(
+        (String) emf.getProperties().get(Environment.ISOLATION));
+  }
+
+  @Override
+  public TransactionIsolationLevel getCurrentTransactionIsolationLevel() {
+    assertInTransaction();
+    String mode =
+        (String)
+            getEntityManager()
+                .createNativeQuery("SHOW TRANSACTION ISOLATION LEVEL")
+                .getSingleResult();
+    return TransactionIsolationLevel.fromMode(mode);
   }
 
   @Override
@@ -287,7 +357,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
                 Integer.class)
             .setMaxResults(1);
     entityIds.forEach(entityId -> query.setParameter(entityId.name, entityId.value));
-    return query.getResultList().size() > 0;
+    return !query.getResultList().isEmpty();
   }
 
   @Override
@@ -506,7 +576,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
         .collect(toImmutableSet());
   }
 
-  private String getAndClause(ImmutableSet<EntityId> entityIds) {
+  private static String getAndClause(ImmutableSet<EntityId> entityIds) {
     return entityIds.stream()
         .map(entityId -> String.format("%s = :%s", entityId.name, entityId.name))
         .collect(joining(" AND "));
@@ -556,7 +626,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
     try {
       getEntityManager().getMetamodel().entity(object.getClass());
     } catch (IllegalArgumentException e) {
-      // The object is not an entity.  Return without detaching.
+      // The object is not an entity. Return without detaching.
       return object;
     }
 
@@ -637,7 +707,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
 
     TypedQuery<T> delegate;
 
-    public DetachingTypedQuery(TypedQuery<T> delegate) {
+    DetachingTypedQuery(TypedQuery<T> delegate) {
       this.delegate = delegate;
     }
 
@@ -866,7 +936,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
     @Override
     public Optional<T> first() {
       List<T> results = buildQuery().setMaxResults(1).getResultList();
-      return results.size() > 0 ? Optional.of(detach(results.get(0))) : Optional.empty();
+      return !results.isEmpty() ? Optional.of(detach(results.get(0))) : Optional.empty();
     }
 
     @Override
@@ -895,7 +965,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
     public ImmutableList<T> list() {
       return buildQuery().getResultList().stream()
           .map(JpaTransactionManagerImpl.this::detach)
-          .collect(ImmutableList.toImmutableList());
+          .collect(toImmutableList());
     }
   }
 }

core/src/main/java/google/registry/persistence/transaction/TransactionManager.java

@@ -18,6 +18,7 @@ import com.google.common.collect.ImmutableCollection;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import google.registry.model.ImmutableObject;
+import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
 import google.registry.persistence.VKey;
 import java.util.NoSuchElementException;
 import java.util.Optional;
@@ -49,9 +50,18 @@ public interface TransactionManager {
   /** Executes the work in a transaction and returns the result. */
   <T> T transact(Supplier<T> work);
 
+  /**
+   * Executes the work in a transaction at the given {@link TransactionIsolationLevel} and returns
+   * the result.
+   */
+  <T> T transact(Supplier<T> work, TransactionIsolationLevel isolationLevel);
+
   /** Executes the work in a transaction. */
   void transact(Runnable work);
 
+  /** Executes the work in a transaction at the given {@link TransactionIsolationLevel}. */
+  void transact(Runnable work, TransactionIsolationLevel isolationLevel);
+
   /** Returns the time associated with the start of this particular transaction attempt. */
   DateTime getTransactionTime();
 

core/src/test/java/google/registry/flows/FlowRunnerTest.java

@@ -17,6 +17,7 @@ package google.registry.flows;
 import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.truth.Truth.assertWithMessage;
 import static com.google.common.truth.Truth8.assertThat;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.testing.TestDataHelper.loadFile;
 import static google.registry.testing.TestLogHandlerUtils.findFirstLogMessageByPrefix;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
@@ -30,11 +31,13 @@ import com.google.common.base.Splitter;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSortedMap;
 import com.google.common.testing.TestLogHandler;
+import google.registry.config.RegistryConfig;
 import google.registry.flows.certs.CertificateChecker;
 import google.registry.model.eppcommon.Trid;
 import google.registry.model.eppoutput.EppOutput.ResponseOrGreeting;
 import google.registry.model.eppoutput.EppResponse;
 import google.registry.monitoring.whitebox.EppMetric;
+import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
 import google.registry.persistence.transaction.JpaTestExtensions;
 import google.registry.persistence.transaction.JpaTestExtensions.JpaIntegrationTestExtension;
 import google.registry.testing.FakeClock;
@@ -46,7 +49,6 @@ import org.joda.time.DateTime;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
-import org.mockito.Mockito;
 
 /** Unit tests for {@link FlowRunner}. */
 class FlowRunnerTest {
@@ -78,6 +80,21 @@ class FlowRunnerTest {
     }
   }
 
+  static class TestTransactionalFlow implements TransactionalFlow {
+    private final Optional<TransactionIsolationLevel> isolationLevel;
+
+    TestTransactionalFlow(Optional<TransactionIsolationLevel> isolationLevel) {
+      this.isolationLevel = isolationLevel;
+    }
+
+    @Override
+    public ResponseOrGreeting run() {
+      tm().assertTransactionIsolationLevel(
+              isolationLevel.orElse(tm().getDefaultTransactionIsolationLevel()));
+      return mock(EppResponse.class);
+    }
+  }
+
   @BeforeEach
   void beforeEach() {
     JdkLoggerConfig.getConfig(FlowRunner.class).addHandler(handler);
@@ -90,23 +107,39 @@ class FlowRunnerTest {
     flowRunner.isDryRun = false;
     flowRunner.isSuperuser = false;
     flowRunner.isTransactional = false;
+    flowRunner.isolationLevelOverride = Optional.empty();
     flowRunner.sessionMetadata =
         new StatelessRequestSessionMetadata("TheRegistrar", ImmutableSet.of());
     flowRunner.trid = Trid.create("client-123", "server-456");
-    flowRunner.flowReporter = Mockito.mock(FlowReporter.class);
+    flowRunner.flowReporter = mock(FlowReporter.class);
   }
 
   @Test
   void testRun_nonTransactionalCommand_setsCommandNameOnMetric() throws Exception {
-    flowRunner.isTransactional = true;
     flowRunner.run(eppMetricBuilder);
     assertThat(eppMetricBuilder.build().getCommandName()).hasValue("TestCommand");
   }
 
   @Test
   void testRun_transactionalCommand_setsCommandNameOnMetric() throws Exception {
+    flowRunner.isTransactional = true;
+    flowRunner.flowClass = TestTransactionalFlow.class;
+    flowRunner.flowProvider = () -> new TestTransactionalFlow(Optional.empty());
     flowRunner.run(eppMetricBuilder);
-    assertThat(eppMetricBuilder.build().getCommandName()).hasValue("TestCommand");
+    assertThat(eppMetricBuilder.build().getCommandName()).hasValue("TestTransactional");
+  }
+
+  @Test
+  void testRun_transactionalCommand_isolationLevelOverride() throws Exception {
+    flowRunner.isTransactional = true;
+    flowRunner.isolationLevelOverride =
+        Optional.of(TransactionIsolationLevel.TRANSACTION_READ_UNCOMMITTED);
+    flowRunner.flowClass = TestTransactionalFlow.class;
+    flowRunner.flowProvider = () -> new TestTransactionalFlow(flowRunner.isolationLevelOverride);
+    if (RegistryConfig.getHibernatePerTransactionIsolationEnabled()) {
+      flowRunner.run(eppMetricBuilder);
+      assertThat(eppMetricBuilder.build().getCommandName()).hasValue("TestTransactional");
+    }
   }
 
   @Test

core/src/test/java/google/registry/persistence/transaction/JpaTransactionManagerImplTest.java

@@ -16,6 +16,8 @@ package google.registry.persistence.transaction;
 
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.common.truth.Truth.assertThat;
+import static google.registry.persistence.PersistenceModule.TransactionIsolationLevel.TRANSACTION_READ_UNCOMMITTED;
+import static google.registry.persistence.PersistenceModule.TransactionIsolationLevel.TRANSACTION_REPEATABLE_READ;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.testing.DatabaseHelper.assertDetachedFromEntityManager;
 import static google.registry.testing.DatabaseHelper.existsInDb;
@@ -31,6 +33,7 @@ import static org.mockito.Mockito.verify;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import google.registry.config.RegistryConfig;
 import google.registry.model.ImmutableObject;
 import google.registry.persistence.VKey;
 import google.registry.persistence.transaction.JpaTestExtensions.JpaUnitTestExtension;
@@ -81,7 +84,7 @@ class JpaTransactionManagerImplTest {
           .buildUnitTestExtension();
 
   @Test
-  void transact_succeeds() {
+  void transact_success() {
     assertPersonEmpty();
     assertCompanyEmpty();
     tm().transact(
@@ -89,6 +92,7 @@ class JpaTransactionManagerImplTest {
               insertPerson(10);
               insertCompany("Foo");
               insertCompany("Bar");
+              tm().assertTransactionIsolationLevel(tm().getDefaultTransactionIsolationLevel());
             });
     assertPersonCount(1);
     assertPersonExist(10);
@@ -97,6 +101,52 @@ class JpaTransactionManagerImplTest {
     assertCompanyExist("Bar");
   }
 
+  @Test
+  void transact_setIsolationLevel() {
+    tm().transact(
+            () -> {
+              tm().assertTransactionIsolationLevel(
+                      RegistryConfig.getHibernatePerTransactionIsolationEnabled()
+                          ? TRANSACTION_READ_UNCOMMITTED
+                          : tm().getDefaultTransactionIsolationLevel());
+              return null;
+            },
+            TRANSACTION_READ_UNCOMMITTED);
+    // Make sure that we can start a new transaction on the same thread with a different isolation
+    // level.
+    tm().transact(
+            () -> {
+              tm().assertTransactionIsolationLevel(
+                      RegistryConfig.getHibernatePerTransactionIsolationEnabled()
+                          ? TRANSACTION_REPEATABLE_READ
+                          : tm().getDefaultTransactionIsolationLevel());
+              return null;
+            },
+            TRANSACTION_REPEATABLE_READ);
+  }
+
+  @Test
+  void transact_nestedTransactions() {
+    // Unit tests always allows for per-transaction isolation level (and therefore throws when a
+    // nested transaction is detected).
+    if (RegistryConfig.getHibernatePerTransactionIsolationEnabled()) {
+      IllegalArgumentException e =
+          assertThrows(
+              IllegalArgumentException.class,
+              () ->
+                  tm().transact(
+                          () -> {
+                            tm().transact(() -> {});
+                          }));
+      assertThat(e).hasMessageThat().isEqualTo("Nested transaction detected");
+    } else {
+      tm().transact(
+              () -> {
+                tm().transact(() -> {});
+              });
+    }
+  }
+
   @Test
   void transact_hasNoEffectWithPartialSuccess() {
     assertPersonEmpty();
@@ -606,19 +656,19 @@ class JpaTransactionManagerImplTest {
     verify(spyJpaTm, times(3)).delete(theEntityKey);
   }
 
-  private void insertPerson(int age) {
+  private static void insertPerson(int age) {
     tm().getEntityManager()
         .createNativeQuery(String.format("INSERT INTO Person (age) VALUES (%d)", age))
         .executeUpdate();
   }
 
-  private void insertCompany(String name) {
+  private static void insertCompany(String name) {
     tm().getEntityManager()
         .createNativeQuery(String.format("INSERT INTO Company (name) VALUES ('%s')", name))
         .executeUpdate();
   }
 
-  private void assertPersonExist(int age) {
+  private static void assertPersonExist(int age) {
     tm().transact(
             () -> {
               EntityManager em = tm().getEntityManager();
@@ -631,7 +681,7 @@ class JpaTransactionManagerImplTest {
             });
   }
 
-  private void assertCompanyExist(String name) {
+  private static void assertCompanyExist(String name) {
     tm().transact(
             () -> {
               String maybeName =
@@ -644,23 +694,23 @@ class JpaTransactionManagerImplTest {
             });
   }
 
-  private void assertPersonCount(int count) {
+  private static void assertPersonCount(int count) {
     assertThat(countTable("Person")).isEqualTo(count);
   }
 
-  private void assertCompanyCount(int count) {
+  private static void assertCompanyCount(int count) {
     assertThat(countTable("Company")).isEqualTo(count);
   }
 
-  private void assertPersonEmpty() {
+  private static void assertPersonEmpty() {
     assertPersonCount(0);
   }
 
-  private void assertCompanyEmpty() {
+  private static void assertCompanyEmpty() {
     assertCompanyCount(0);
   }
 
-  private int countTable(String tableName) {
+  private static int countTable(String tableName) {
     return tm().transact(
             () -> {
               BigInteger colCount =

core/src/test/java/google/registry/persistence/transaction/ReplicaSimulatingJpaTransactionManager.java

@@ -18,6 +18,7 @@ import com.google.common.collect.ImmutableCollection;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import google.registry.model.ImmutableObject;
+import google.registry.persistence.PersistenceModule.TransactionIsolationLevel;
 import google.registry.persistence.VKey;
 import java.util.Optional;
 import java.util.function.Supplier;
@@ -49,6 +50,21 @@ public class ReplicaSimulatingJpaTransactionManager implements JpaTransactionMan
     delegate.teardown();
   }
 
+  @Override
+  public TransactionIsolationLevel getDefaultTransactionIsolationLevel() {
+    return delegate.getDefaultTransactionIsolationLevel();
+  }
+
+  @Override
+  public TransactionIsolationLevel getCurrentTransactionIsolationLevel() {
+    return delegate.getCurrentTransactionIsolationLevel();
+  }
+
+  @Override
+  public void assertTransactionIsolationLevel(TransactionIsolationLevel expectedLevel) {
+    delegate.assertTransactionIsolationLevel(expectedLevel);
+  }
+
   @Override
   public EntityManager getStandaloneEntityManager() {
     return delegate.getStandaloneEntityManager();
@@ -85,7 +101,7 @@ public class ReplicaSimulatingJpaTransactionManager implements JpaTransactionMan
   }
 
   @Override
-  public <T> T transact(Supplier<T> work) {
+  public <T> T transact(Supplier<T> work, TransactionIsolationLevel isolationLevel) {
     if (delegate.inTransaction()) {
       return work.get();
     }
@@ -96,26 +112,48 @@ public class ReplicaSimulatingJpaTransactionManager implements JpaTransactionMan
               .createNativeQuery("SET TRANSACTION READ ONLY")
               .executeUpdate();
           return work.get();
-        });
+        },
+        isolationLevel);
+  }
+
+  @Override
+  public <T> T transact(Supplier<T> work) {
+    return transact(work, null);
+  }
+
+  @Override
+  public <T> T transactNoRetry(Supplier<T> work, TransactionIsolationLevel isolationLevel) {
+    return transact(work, isolationLevel);
   }
 
   @Override
   public <T> T transactNoRetry(Supplier<T> work) {
-    return transact(work);
+    return transactNoRetry(work, null);
   }
 
   @Override
-  public void transact(Runnable work) {
+  public void transact(Runnable work, TransactionIsolationLevel isolationLevel) {
     transact(
         () -> {
           work.run();
           return null;
-        });
+        },
+        isolationLevel);
+  }
+
+  @Override
+  public void transact(Runnable work) {
+    transact(work, null);
+  }
+
+  @Override
+  public void transactNoRetry(Runnable work, TransactionIsolationLevel isolationLevel) {
+    transact(work, isolationLevel);
   }
 
   @Override
   public void transactNoRetry(Runnable work) {
-    transact(work);
+    transactNoRetry(work, null);
   }
 
   @Override