Fix flaky RdeStagingActionDatastoreTest (#1514 )

* Fix flaky RdeStagingActionDatastoreTest Fixed the most common cause that makes one method flaky (Clock and timestamp problem). Added a TODO to rethink test case. Also added notes on tasks potentially enqueued multiple times.
Add support for delay of duration when scheduling a task (#1493 )
2026-06-09 16:33:02 +00:00 · 2022-02-04 10:40:52 -05:00 · 2022-02-03 22:25:39 -05:00 · 2022-02-03 16:04:58 -05:00 · 2022-02-03 15:57:15 -05:00 · 2022-02-03 11:14:26 -05:00
45 changed files with 1441 additions and 831 deletions
@@ -104,6 +104,7 @@ final class ValidateSqlUtils {
    private final HashMap<String, Counter> missingCounters = new HashMap<>();
    private final HashMap<String, Counter> unequalCounters = new HashMap<>();
    private final HashMap<String, Counter> badEntityCounters = new HashMap<>();
+    private final HashMap<String, Counter> duplicateEntityCounters = new HashMap<>();

    private volatile boolean logPrinted = false;

@@ -120,6 +121,8 @@ final class ValidateSqlUtils {
          counterKey, Metrics.counter("CompareDB", "Missing In One DB: " + counterKey));
      unequalCounters.put(counterKey, Metrics.counter("CompareDB", "Not Equal:" + counterKey));
      badEntityCounters.put(counterKey, Metrics.counter("CompareDB", "Bad Entities:" + counterKey));
+      duplicateEntityCounters.put(
+          counterKey, Metrics.counter("CompareDB", "Duplicate Entities:" + counterKey));
    }

    /**
@@ -158,12 +161,18 @@ final class ValidateSqlUtils {
      ImmutableList<SqlEntity> entities = ImmutableList.copyOf(kv.getValue());

      verify(!entities.isEmpty(), "Can't happen: no value for key %s.", kv.getKey());
-      verify(entities.size() <= 2, "Unexpected duplicates for key %s", kv.getKey());

      String counterKey = getCounterKey(entities.get(0).getClass());
      ensureCounterExists(counterKey);
      totalCounters.get(counterKey).inc();

+      if (entities.size() > 2) {
+        // Duplicates may happen with Cursors if imported across projects. Its key in Datastore, the
+        // id field, encodes the project name and is not fixed by the importing job.
+        duplicateEntityCounters.get(counterKey).inc();
+        return;
+      }
+
      if (entities.size() == 1) {
        if (isSpecialCaseProberEntity(entities.get(0))) {
          return;
@@ -297,6 +297,10 @@ public class RdeIO {
                logger.atInfo().log(
                    "Rolled forward %s on %s cursor to %s.", key.cursor(), key.tld(), newPosition);
                RdeRevision.saveRevision(key.tld(), key.watermark(), key.mode(), revision);
+                // Enqueueing a task is a side effect that is not undone if the transaction rolls
+                // back. So this may result in multiple copies of the same task being processed.
+                // This is fine because the RdeUploadAction is guarded by a lock and tracks progress
+                // by cursor. The BrdaCopyAction writes a file to GCS, which is an atomic action.
                if (key.mode() == RdeMode.FULL) {
                  cloudTasksUtils.enqueue(
                      RDE_UPLOAD_QUEUE,
@@ -169,6 +169,7 @@ import org.joda.time.Duration;
 * @error {@link DomainFlowUtils.FeesMismatchException}
 * @error {@link DomainFlowUtils.FeesRequiredDuringEarlyAccessProgramException}
 * @error {@link DomainFlowUtils.FeesRequiredForPremiumNameException}
+ * @error {@link DomainFlowUtils.InvalidDsRecordException}
 * @error {@link DomainFlowUtils.InvalidIdnDomainLabelException}
 * @error {@link DomainFlowUtils.InvalidPunycodeException}
 * @error {@link DomainFlowUtils.InvalidTcnIdChecksumException}
@@ -129,6 +129,7 @@ import google.registry.model.tld.label.ReservedList;
 import google.registry.model.tmch.ClaimsListDao;
 import google.registry.persistence.VKey;
 import google.registry.tldconfig.idn.IdnLabelValidator;
+import google.registry.tools.DigestType;
 import google.registry.util.Idn;
 import java.math.BigDecimal;
 import java.util.Collection;
@@ -144,6 +145,7 @@ import org.joda.money.CurrencyUnit;
 import org.joda.money.Money;
 import org.joda.time.DateTime;
 import org.joda.time.Duration;
+import org.xbill.DNS.DNSSEC.Algorithm;

 /** Static utility functions for domain flows. */
 public class DomainFlowUtils {
@@ -293,13 +295,46 @@ public class DomainFlowUtils {

  /** Check that the DS data that will be set on a domain is valid. */
  static void validateDsData(Set<DelegationSignerData> dsData) throws EppException {
-    if (dsData != null && dsData.size() > MAX_DS_RECORDS_PER_DOMAIN) {
-      throw new TooManyDsRecordsException(
-          String.format(
-              "A maximum of %s DS records are allowed per domain.", MAX_DS_RECORDS_PER_DOMAIN));
+    if (dsData != null) {
+      if (dsData.size() > MAX_DS_RECORDS_PER_DOMAIN) {
+        throw new TooManyDsRecordsException(
+            String.format(
+                "A maximum of %s DS records are allowed per domain.", MAX_DS_RECORDS_PER_DOMAIN));
+      }
+      // TODO(sarahbot@): Add signature length verification
+      ImmutableList<DelegationSignerData> invalidAlgorithms =
+          dsData.stream()
+              .filter(ds -> !validateAlgorithm(ds.getAlgorithm()))
+              .collect(toImmutableList());
+      if (!invalidAlgorithms.isEmpty()) {
+        throw new InvalidDsRecordException(
+            String.format(
+                "Domain contains DS record(s) with an invalid algorithm wire value: %s",
+                invalidAlgorithms));
+      }
+      ImmutableList<DelegationSignerData> invalidDigestTypes =
+          dsData.stream()
+              .filter(ds -> !DigestType.fromWireValue(ds.getDigestType()).isPresent())
+              .collect(toImmutableList());
+      if (!invalidDigestTypes.isEmpty()) {
+        throw new InvalidDsRecordException(
+            String.format(
+                "Domain contains DS record(s) with an invalid digest type: %s",
+                invalidDigestTypes));
+      }
    }
  }

+  public static boolean validateAlgorithm(int alg) {
+    if (alg > 255 || alg < 0) {
+      return false;
+    }
+    // Algorithms that are reserved or unassigned will just return a string representation of their
+    // integer wire value.
+    String algorithm = Algorithm.string(alg);
+    return !algorithm.equals(Integer.toString(alg));
+  }
+
  /** We only allow specifying years in a period. */
  static Period verifyUnitIsYears(Period period) throws EppException {
    if (!checkNotNull(period).getUnit().equals(Period.Unit.YEARS)) {
@@ -1217,6 +1252,13 @@ public class DomainFlowUtils {
    }
  }

+  /** Domain has an invalid DS record. */
+  static class InvalidDsRecordException extends ParameterValuePolicyErrorException {
+    public InvalidDsRecordException(String message) {
+      super(message);
+    }
+  }
+
  /** Domain name is under tld which doesn't exist. */
  static class TldDoesNotExistException extends ParameterValueRangeErrorException {
    public TldDoesNotExistException(String tld) {
@@ -114,6 +114,7 @@ import org.joda.time.DateTime;
 * @error {@link DomainFlowUtils.EmptySecDnsUpdateException}
 * @error {@link DomainFlowUtils.FeesMismatchException}
 * @error {@link DomainFlowUtils.FeesRequiredForNonFreeOperationException}
+ * @error {@link DomainFlowUtils.InvalidDsRecordException}
 * @error {@link DomainFlowUtils.LinkedResourcesDoNotExistException}
 * @error {@link DomainFlowUtils.LinkedResourceInPendingDeleteProhibitsOperationException}
 * @error {@link DomainFlowUtils.MaxSigLifeChangeNotSupportedException}
@@ -34,6 +34,9 @@ import javax.persistence.AccessType;
@ReportedOn
@Entity
@javax.persistence.Entity(name = "Host")
+@javax.persistence.Table(
+    name = "Host",
+    indexes = {@javax.persistence.Index(columnList = "hostName")})
@ExternalMessagingName("host")
@WithStringVKey
@Access(AccessType.FIELD) // otherwise it'll use the default if the repoId (property)
@@ -18,7 +18,6 @@ import static google.registry.persistence.transaction.TransactionManagerFactory.

 import com.google.common.collect.ImmutableList;
 import java.util.Collection;
-import javax.persistence.EntityManager;
 import javax.persistence.criteria.CriteriaBuilder;
 import javax.persistence.criteria.CriteriaQuery;
 import javax.persistence.criteria.Expression;
@@ -42,12 +41,14 @@ public class CriteriaQueryBuilder<T> {

  private final CriteriaQuery<T> query;
  private final Root<?> root;
+  private final JpaTransactionManager jpaTm;
  private final ImmutableList.Builder<Predicate> predicates = new ImmutableList.Builder<>();
  private final ImmutableList.Builder<Order> orders = new ImmutableList.Builder<>();

-  private CriteriaQueryBuilder(CriteriaQuery<T> query, Root<?> root) {
+  private CriteriaQueryBuilder(CriteriaQuery<T> query, Root<?> root, JpaTransactionManager jpaTm) {
    this.query = query;
    this.root = root;
+    this.jpaTm = jpaTm;
  }

  /** Adds a WHERE clause to the query, given the specified operation, field, and value. */
@@ -75,18 +76,18 @@ public class CriteriaQueryBuilder<T> {
   */
  public <V> CriteriaQueryBuilder<T> whereFieldContains(String fieldName, Object value) {
    return where(
-        jpaTm().getEntityManager().getCriteriaBuilder().isMember(value, root.get(fieldName)));
+        jpaTm.getEntityManager().getCriteriaBuilder().isMember(value, root.get(fieldName)));
  }

  /** Orders the result by the given field ascending. */
  public CriteriaQueryBuilder<T> orderByAsc(String fieldName) {
-    orders.add(jpaTm().getEntityManager().getCriteriaBuilder().asc(root.get(fieldName)));
+    orders.add(jpaTm.getEntityManager().getCriteriaBuilder().asc(root.get(fieldName)));
    return this;
  }

  /** Orders the result by the given field descending. */
  public CriteriaQueryBuilder<T> orderByDesc(String fieldName) {
-    orders.add(jpaTm().getEntityManager().getCriteriaBuilder().desc(root.get(fieldName)));
+    orders.add(jpaTm.getEntityManager().getCriteriaBuilder().desc(root.get(fieldName)));
    return this;
  }

@@ -103,23 +104,24 @@ public class CriteriaQueryBuilder<T> {

  /** Creates a query builder that will SELECT from the given class. */
  public static <T> CriteriaQueryBuilder<T> create(Class<T> clazz) {
-    return create(jpaTm().getEntityManager(), clazz);
+    return create(jpaTm(), clazz);
  }

  /** Creates a query builder for the given entity manager. */
-  public static <T> CriteriaQueryBuilder<T> create(EntityManager em, Class<T> clazz) {
-    CriteriaQuery<T> query = em.getCriteriaBuilder().createQuery(clazz);
+  public static <T> CriteriaQueryBuilder<T> create(JpaTransactionManager jpaTm, Class<T> clazz) {
+    CriteriaQuery<T> query = jpaTm.getEntityManager().getCriteriaBuilder().createQuery(clazz);
    Root<T> root = query.from(clazz);
    query = query.select(root);
-    return new CriteriaQueryBuilder<>(query, root);
+    return new CriteriaQueryBuilder<>(query, root, jpaTm);
  }

  /** Creates a "count" query for the table for the class. */
-  public static <T> CriteriaQueryBuilder<Long> createCount(EntityManager em, Class<T> clazz) {
-    CriteriaBuilder builder = em.getCriteriaBuilder();
+  public static <T> CriteriaQueryBuilder<Long> createCount(
+      JpaTransactionManager jpaTm, Class<T> clazz) {
+    CriteriaBuilder builder = jpaTm.getEntityManager().getCriteriaBuilder();
    CriteriaQuery<Long> query = builder.createQuery(Long.class);
    Root<T> root = query.from(clazz);
    query = query.select(builder.count(root));
-    return new CriteriaQueryBuilder<>(query, root);
+    return new CriteriaQueryBuilder<>(query, root, jpaTm);
  }
 }
@@ -1131,7 +1131,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {

    private TypedQuery<T> buildQuery() {
      CriteriaQueryBuilder<T> queryBuilder =
-          CriteriaQueryBuilder.create(getEntityManager(), entityClass);
+          CriteriaQueryBuilder.create(JpaTransactionManagerImpl.this, entityClass);
      return addCriteria(queryBuilder);
    }

@@ -1178,7 +1178,7 @@ public class JpaTransactionManagerImpl implements JpaTransactionManager {
    @Override
    public long count() {
      CriteriaQueryBuilder<Long> queryBuilder =
-          CriteriaQueryBuilder.createCount(getEntityManager(), entityClass);
+          CriteriaQueryBuilder.createCount(JpaTransactionManagerImpl.this, entityClass);
      return addCriteria(queryBuilder).getSingleResult();
    }

@@ -18,6 +18,7 @@ import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static google.registry.model.EppResourceUtils.loadByForeignKey;
 import static google.registry.model.index.ForeignKeyIndex.loadAndGetKey;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
+import static google.registry.persistence.transaction.TransactionManagerFactory.replicaJpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.request.Action.Method.GET;
 import static google.registry.request.Action.Method.HEAD;
@@ -37,10 +38,8 @@ import com.google.common.primitives.Booleans;
 import com.googlecode.objectify.cmd.Query;
 import google.registry.model.domain.DomainBase;
 import google.registry.model.host.HostResource;
-import google.registry.persistence.PersistenceModule.ReadOnlyReplicaJpaTm;
 import google.registry.persistence.VKey;
 import google.registry.persistence.transaction.CriteriaQueryBuilder;
-import google.registry.persistence.transaction.JpaTransactionManager;
 import google.registry.rdap.RdapJsonFormatter.OutputDataType;
 import google.registry.rdap.RdapMetrics.EndpointType;
 import google.registry.rdap.RdapMetrics.SearchType;
@@ -93,8 +92,6 @@ public class RdapDomainSearchAction extends RdapSearchActionBase {
  @Inject @Parameter("nsLdhName") Optional<String> nsLdhNameParam;
  @Inject @Parameter("nsIp") Optional<String> nsIpParam;

-  @Inject @ReadOnlyReplicaJpaTm JpaTransactionManager readOnlyJpaTm;
-
  @Inject
  public RdapDomainSearchAction() {
    super("domain search", EndpointType.DOMAINS);
@@ -228,31 +225,32 @@ public class RdapDomainSearchAction extends RdapSearchActionBase {
      resultSet = getMatchingResources(query, true, querySizeLimit);
    } else {
      resultSet =
-          readOnlyJpaTm.transact(
-              () -> {
-                CriteriaBuilder criteriaBuilder =
-                    readOnlyJpaTm.getEntityManager().getCriteriaBuilder();
-                CriteriaQueryBuilder<DomainBase> queryBuilder =
-                    CriteriaQueryBuilder.create(DomainBase.class)
-                        .where(
-                            "fullyQualifiedDomainName",
-                            criteriaBuilder::like,
-                            String.format("%s%%", partialStringQuery.getInitialString()))
-                        .orderByAsc("fullyQualifiedDomainName");
-                if (cursorString.isPresent()) {
-                  queryBuilder =
-                      queryBuilder.where(
-                          "fullyQualifiedDomainName",
-                          criteriaBuilder::greaterThan,
-                          cursorString.get());
-                }
-                if (partialStringQuery.getSuffix() != null) {
-                  queryBuilder =
-                      queryBuilder.where(
-                          "tld", criteriaBuilder::equal, partialStringQuery.getSuffix());
-                }
-                return getMatchingResourcesSql(queryBuilder, true, querySizeLimit);
-              });
+          replicaJpaTm()
+              .transact(
+                  () -> {
+                    CriteriaBuilder criteriaBuilder =
+                        replicaJpaTm().getEntityManager().getCriteriaBuilder();
+                    CriteriaQueryBuilder<DomainBase> queryBuilder =
+                        CriteriaQueryBuilder.create(replicaJpaTm(), DomainBase.class)
+                            .where(
+                                "fullyQualifiedDomainName",
+                                criteriaBuilder::like,
+                                String.format("%s%%", partialStringQuery.getInitialString()))
+                            .orderByAsc("fullyQualifiedDomainName");
+                    if (cursorString.isPresent()) {
+                      queryBuilder =
+                          queryBuilder.where(
+                              "fullyQualifiedDomainName",
+                              criteriaBuilder::greaterThan,
+                              cursorString.get());
+                    }
+                    if (partialStringQuery.getSuffix() != null) {
+                      queryBuilder =
+                          queryBuilder.where(
+                              "tld", criteriaBuilder::equal, partialStringQuery.getSuffix());
+                    }
+                    return getMatchingResourcesSql(queryBuilder, true, querySizeLimit);
+                  });
    }
    return makeSearchResults(resultSet);
  }
@@ -274,19 +272,20 @@ public class RdapDomainSearchAction extends RdapSearchActionBase {
      resultSet = getMatchingResources(query, true, querySizeLimit);
    } else {
      resultSet =
-          readOnlyJpaTm.transact(
-              () -> {
-                CriteriaQueryBuilder<DomainBase> builder =
-                    queryItemsSql(
-                            DomainBase.class,
-                            "tld",
-                            tld,
-                            Optional.of("fullyQualifiedDomainName"),
-                            cursorString,
-                            DeletedItemHandling.INCLUDE)
-                        .orderByAsc("fullyQualifiedDomainName");
-                return getMatchingResourcesSql(builder, true, querySizeLimit);
-              });
+          replicaJpaTm()
+              .transact(
+                  () -> {
+                    CriteriaQueryBuilder<DomainBase> builder =
+                        queryItemsSql(
+                                DomainBase.class,
+                                "tld",
+                                tld,
+                                Optional.of("fullyQualifiedDomainName"),
+                                cursorString,
+                                DeletedItemHandling.INCLUDE)
+                            .orderByAsc("fullyQualifiedDomainName");
+                    return getMatchingResourcesSql(builder, true, querySizeLimit);
+                  });
    }
    return makeSearchResults(resultSet);
  }
@@ -357,28 +356,29 @@ public class RdapDomainSearchAction extends RdapSearchActionBase {
          .map(VKey::from)
          .collect(toImmutableSet());
    } else {
-      return readOnlyJpaTm.transact(
-          () -> {
-            CriteriaQueryBuilder<HostResource> builder =
-                queryItemsSql(
-                    HostResource.class,
-                    "fullyQualifiedHostName",
-                    partialStringQuery,
-                    Optional.empty(),
-                    DeletedItemHandling.EXCLUDE);
-            if (desiredRegistrar.isPresent()) {
-              builder =
-                  builder.where(
-                      "currentSponsorClientId",
-                      readOnlyJpaTm.getEntityManager().getCriteriaBuilder()::equal,
-                      desiredRegistrar.get());
-            }
-            return getMatchingResourcesSql(builder, true, maxNameserversInFirstStage)
-                .resources()
-                .stream()
-                .map(HostResource::createVKey)
-                .collect(toImmutableSet());
-          });
+      return replicaJpaTm()
+          .transact(
+              () -> {
+                CriteriaQueryBuilder<HostResource> builder =
+                    queryItemsSql(
+                        HostResource.class,
+                        "fullyQualifiedHostName",
+                        partialStringQuery,
+                        Optional.empty(),
+                        DeletedItemHandling.EXCLUDE);
+                if (desiredRegistrar.isPresent()) {
+                  builder =
+                      builder.where(
+                          "currentSponsorClientId",
+                          replicaJpaTm().getEntityManager().getCriteriaBuilder()::equal,
+                          desiredRegistrar.get());
+                }
+                return getMatchingResourcesSql(builder, true, maxNameserversInFirstStage)
+                    .resources()
+                    .stream()
+                    .map(HostResource::createVKey)
+                    .collect(toImmutableSet());
+              });
    }
  }

@@ -512,20 +512,21 @@ public class RdapDomainSearchAction extends RdapSearchActionBase {
        parameters.put("desiredRegistrar", desiredRegistrar.get());
      }
      hostKeys =
-          readOnlyJpaTm.transact(
-              () -> {
-                javax.persistence.Query query =
-                    readOnlyJpaTm
-                        .getEntityManager()
-                        .createNativeQuery(queryBuilder.toString())
-                        .setMaxResults(maxNameserversInFirstStage);
-                parameters.build().forEach(query::setParameter);
-                @SuppressWarnings("unchecked")
-                Stream<String> resultStream = query.getResultStream();
-                return resultStream
-                    .map(repoId -> VKey.create(HostResource.class, repoId))
-                    .collect(toImmutableSet());
-              });
+          replicaJpaTm()
+              .transact(
+                  () -> {
+                    javax.persistence.Query query =
+                        replicaJpaTm()
+                            .getEntityManager()
+                            .createNativeQuery(queryBuilder.toString())
+                            .setMaxResults(maxNameserversInFirstStage);
+                    parameters.build().forEach(query::setParameter);
+                    @SuppressWarnings("unchecked")
+                    Stream<String> resultStream = query.getResultStream();
+                    return resultStream
+                        .map(repoId -> VKey.create(HostResource.class, repoId))
+                        .collect(toImmutableSet());
+                  });
    }
    return searchByNameserverRefs(hostKeys);
  }
@@ -570,38 +571,39 @@ public class RdapDomainSearchAction extends RdapSearchActionBase {
        }
        stream.forEach(domainSetBuilder::add);
      } else {
-        readOnlyJpaTm.transact(
-            () -> {
-              for (VKey<HostResource> hostKey : hostKeys) {
-                CriteriaQueryBuilder<DomainBase> queryBuilder =
-                    CriteriaQueryBuilder.create(DomainBase.class)
-                        .whereFieldContains("nsHosts", hostKey)
-                        .orderByAsc("fullyQualifiedDomainName");
-                CriteriaBuilder criteriaBuilder =
-                    readOnlyJpaTm.getEntityManager().getCriteriaBuilder();
-                if (!shouldIncludeDeleted()) {
-                  queryBuilder =
-                      queryBuilder.where(
-                          "deletionTime", criteriaBuilder::greaterThan, getRequestTime());
-                }
-                if (cursorString.isPresent()) {
-                  queryBuilder =
-                      queryBuilder.where(
-                          "fullyQualifiedDomainName",
-                          criteriaBuilder::greaterThan,
-                          cursorString.get());
-                }
-                readOnlyJpaTm
-                    .criteriaQuery(queryBuilder.build())
-                    .getResultStream()
-                    .filter(this::isAuthorized)
-                    .forEach(
-                        (domain) -> {
-                          Hibernate.initialize(domain.getDsData());
-                          domainSetBuilder.add(domain);
-                        });
-              }
-            });
+        replicaJpaTm()
+            .transact(
+                () -> {
+                  for (VKey<HostResource> hostKey : hostKeys) {
+                    CriteriaQueryBuilder<DomainBase> queryBuilder =
+                        CriteriaQueryBuilder.create(replicaJpaTm(), DomainBase.class)
+                            .whereFieldContains("nsHosts", hostKey)
+                            .orderByAsc("fullyQualifiedDomainName");
+                    CriteriaBuilder criteriaBuilder =
+                        replicaJpaTm().getEntityManager().getCriteriaBuilder();
+                    if (!shouldIncludeDeleted()) {
+                      queryBuilder =
+                          queryBuilder.where(
+                              "deletionTime", criteriaBuilder::greaterThan, getRequestTime());
+                    }
+                    if (cursorString.isPresent()) {
+                      queryBuilder =
+                          queryBuilder.where(
+                              "fullyQualifiedDomainName",
+                              criteriaBuilder::greaterThan,
+                              cursorString.get());
+                    }
+                    replicaJpaTm()
+                        .criteriaQuery(queryBuilder.build())
+                        .getResultStream()
+                        .filter(this::isAuthorized)
+                        .forEach(
+                            (domain) -> {
+                              Hibernate.initialize(domain.getDsData());
+                              domainSetBuilder.add(domain);
+                            });
+                  }
+                });
      }
    }
    List<DomainBase> domains = domainSetBuilder.build().asList();
@@ -15,7 +15,7 @@
 package google.registry.rdap;

 import static com.google.common.collect.ImmutableList.toImmutableList;
-import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.replicaJpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.persistence.transaction.TransactionManagerUtil.transactIfJpaTm;
 import static google.registry.rdap.RdapUtils.getRegistrarByIanaIdentifier;
@@ -277,7 +277,7 @@ public class RdapEntitySearchAction extends RdapSearchActionBase {
          resultSet = getMatchingResources(query, false, rdapResultSetMaxSize + 1);
        } else {
          resultSet =
-              jpaTm()
+              replicaJpaTm()
                  .transact(
                      () -> {
                        CriteriaQueryBuilder<ContactResource> builder =
@@ -399,7 +399,7 @@ public class RdapEntitySearchAction extends RdapSearchActionBase {
                  querySizeLimit);
        } else {
          contactResultSet =
-              jpaTm()
+              replicaJpaTm()
                  .transact(
                      () ->
                          getMatchingResourcesSql(
@@ -15,7 +15,7 @@
 package google.registry.rdap;

 import static google.registry.model.EppResourceUtils.loadByForeignKey;
-import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.replicaJpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.request.Action.Method.GET;
 import static google.registry.request.Action.Method.HEAD;
@@ -233,7 +233,7 @@ public class RdapNameserverSearchAction extends RdapSearchActionBase {
      return makeSearchResults(
          getMatchingResources(query, shouldIncludeDeleted(), querySizeLimit), CursorType.NAME);
    } else {
-      return jpaTm()
+      return replicaJpaTm()
          .transact(
              () -> {
                CriteriaQueryBuilder<HostResource> queryBuilder =
@@ -290,11 +290,11 @@ public class RdapNameserverSearchAction extends RdapSearchActionBase {
      }
      queryBuilder.append(" ORDER BY repo_id ASC");
      rdapResultSet =
-          jpaTm()
+          replicaJpaTm()
              .transact(
                  () -> {
                    javax.persistence.Query query =
-                        jpaTm()
+                        replicaJpaTm()
                            .getEntityManager()
                            .createNativeQuery(queryBuilder.toString(), HostResource.class)
                            .setMaxResults(querySizeLimit);
@@ -16,7 +16,7 @@ package google.registry.rdap;

 import static com.google.common.base.Charsets.UTF_8;
 import static google.registry.model.ofy.ObjectifyService.auditedOfy;
-import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.replicaJpaTm;
 import static google.registry.util.DateTimeUtils.END_OF_TIME;

 import com.google.common.collect.ImmutableList;
@@ -193,16 +193,17 @@ public abstract class RdapSearchActionBase extends RdapActionBase {
   */
  <T extends EppResource> RdapResultSet<T> getMatchingResourcesSql(
      CriteriaQueryBuilder<T> builder, boolean checkForVisibility, int querySizeLimit) {
-    jpaTm().assertInTransaction();
+    replicaJpaTm().assertInTransaction();
    Optional<String> desiredRegistrar = getDesiredRegistrar();
    if (desiredRegistrar.isPresent()) {
      builder =
          builder.where(
-              "currentSponsorClientId", jpaTm().getEntityManager().getCriteriaBuilder()::equal,
+              "currentSponsorClientId",
+              replicaJpaTm().getEntityManager().getCriteriaBuilder()::equal,
              desiredRegistrar.get());
    }
    List<T> queryResult =
-        jpaTm().criteriaQuery(builder.build()).setMaxResults(querySizeLimit).getResultList();
+        replicaJpaTm().criteriaQuery(builder.build()).setMaxResults(querySizeLimit).getResultList();
    if (checkForVisibility) {
      return filterResourcesByVisibility(queryResult, querySizeLimit);
    } else {
@@ -395,7 +396,7 @@ public abstract class RdapSearchActionBase extends RdapActionBase {
      RdapSearchPattern partialStringQuery,
      Optional<String> cursorString,
      DeletedItemHandling deletedItemHandling) {
-    jpaTm().assertInTransaction();
+    replicaJpaTm().assertInTransaction();
    if (partialStringQuery.getInitialString().length()
        < RdapSearchPattern.MIN_INITIAL_STRING_LENGTH) {
      throw new UnprocessableEntityException(
@@ -403,8 +404,8 @@ public abstract class RdapSearchActionBase extends RdapActionBase {
              "Initial search string must be at least %d characters",
              RdapSearchPattern.MIN_INITIAL_STRING_LENGTH));
    }
-    CriteriaBuilder criteriaBuilder = jpaTm().getEntityManager().getCriteriaBuilder();
-    CriteriaQueryBuilder<T> builder = CriteriaQueryBuilder.create(clazz);
+    CriteriaBuilder criteriaBuilder = replicaJpaTm().getEntityManager().getCriteriaBuilder();
+    CriteriaQueryBuilder<T> builder = CriteriaQueryBuilder.create(replicaJpaTm(), clazz);
    if (partialStringQuery.getHasWildcard()) {
      builder =
          builder.where(
@@ -493,9 +494,9 @@ public abstract class RdapSearchActionBase extends RdapActionBase {
              "Initial search string must be at least %d characters",
              RdapSearchPattern.MIN_INITIAL_STRING_LENGTH));
    }
-    jpaTm().assertInTransaction();
-    CriteriaQueryBuilder<T> builder = CriteriaQueryBuilder.create(clazz);
-    CriteriaBuilder criteriaBuilder = jpaTm().getEntityManager().getCriteriaBuilder();
+    replicaJpaTm().assertInTransaction();
+    CriteriaQueryBuilder<T> builder = CriteriaQueryBuilder.create(replicaJpaTm(), clazz);
+    CriteriaBuilder criteriaBuilder = replicaJpaTm().getEntityManager().getCriteriaBuilder();
    builder = builder.where(filterField, criteriaBuilder::equal, queryString);
    if (cursorString.isPresent()) {
      if (cursorField.isPresent()) {
@@ -544,7 +545,7 @@ public abstract class RdapSearchActionBase extends RdapActionBase {
      RdapSearchPattern partialStringQuery,
      Optional<String> cursorString,
      DeletedItemHandling deletedItemHandling) {
-    jpaTm().assertInTransaction();
+    replicaJpaTm().assertInTransaction();
    return queryItemsSql(clazz, "repoId", partialStringQuery, cursorString, deletedItemHandling);
  }

@@ -553,7 +554,9 @@ public abstract class RdapSearchActionBase extends RdapActionBase {
    if (!Objects.equals(deletedItemHandling, DeletedItemHandling.INCLUDE)) {
      builder =
          builder.where(
-              "deletionTime", jpaTm().getEntityManager().getCriteriaBuilder()::equal, END_OF_TIME);
+              "deletionTime",
+              replicaJpaTm().getEntityManager().getCriteriaBuilder()::equal,
+              END_OF_TIME);
    }
    return builder;
  }
@@ -24,6 +24,7 @@ import google.registry.config.RegistryConfig.Config;
 import google.registry.gcs.GcsUtils;
 import google.registry.keyring.api.KeyModule.Key;
 import google.registry.model.rde.RdeNamingUtils;
+import google.registry.model.rde.RdeRevision;
 import google.registry.request.Action;
 import google.registry.request.Parameter;
 import google.registry.request.RequestParameters;
@@ -86,7 +87,13 @@ public final class BrdaCopyAction implements Runnable {
  }

  private void copyAsRyde() throws IOException {
-    String nameWithoutPrefix = RdeNamingUtils.makeRydeFilename(tld, watermark, THIN, 1, 0);
+    // TODO(b/217772483): consider guarding this action with a lock and check if there is work.
+    // Not urgent since file writes on GCS are atomic.
+    int revision =
+        RdeRevision.getCurrentRevision(tld, watermark, THIN)
+            .orElseThrow(
+                () -> new IllegalStateException("RdeRevision was not set on generated deposit"));
+    String nameWithoutPrefix = RdeNamingUtils.makeRydeFilename(tld, watermark, THIN, 1, revision);
    String name = prefix.orElse("") + nameWithoutPrefix;
    BlobId xmlFilename = BlobId.of(stagingBucket, name + ".xml.ghostryde");
    BlobId xmlLengthFilename = BlobId.of(stagingBucket, name + ".xml.length");
@@ -391,9 +391,6 @@ public final class RdeStagingAction implements Runnable {
    if (revision.isPresent()) {
      throw new BadRequestException("Revision parameter not allowed in standard operation");
    }
-    if (beam) {
-      throw new BadRequestException("Beam parameter not allowed in standard operation");
-    }

    return ImmutableSetMultimap.copyOf(
        Multimaps.filterValues(
@@ -14,8 +14,6 @@

 package google.registry.rde;

-import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
-import static com.google.appengine.api.taskqueue.TaskOptions.Builder.withUrl;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.base.Verify.verify;
 import static google.registry.model.common.Cursor.getCursorTimeOrStartOfTime;
@@ -26,6 +24,7 @@ import static java.nio.charset.StandardCharsets.UTF_8;
 import com.google.appengine.tools.mapreduce.Reducer;
 import com.google.appengine.tools.mapreduce.ReducerInput;
 import com.google.cloud.storage.BlobId;
+import com.google.common.collect.ImmutableMultimap;
 import com.google.common.flogger.FluentLogger;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.gcs.GcsUtils;
@@ -36,10 +35,11 @@ import google.registry.model.rde.RdeMode;
 import google.registry.model.rde.RdeNamingUtils;
 import google.registry.model.rde.RdeRevision;
 import google.registry.model.tld.Registry;
+import google.registry.request.Action.Service;
 import google.registry.request.RequestParameters;
 import google.registry.request.lock.LockHandler;
 import google.registry.tldconfig.idn.IdnTableEnum;
-import google.registry.util.TaskQueueUtils;
+import google.registry.util.CloudTasksUtils;
 import google.registry.xjc.rdeheader.XjcRdeHeader;
 import google.registry.xjc.rdeheader.XjcRdeHeaderElement;
 import google.registry.xml.ValidationMode;
@@ -65,7 +65,7 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag

  private static final FluentLogger logger = FluentLogger.forEnclosingClass();

-  private final TaskQueueUtils taskQueueUtils;
+  private final CloudTasksUtils cloudTasksUtils;
  private final LockHandler lockHandler;
  private final String bucket;
  private final Duration lockTimeout;
@@ -74,14 +74,14 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
  private final GcsUtils gcsUtils;

  RdeStagingReducer(
-      TaskQueueUtils taskQueueUtils,
+      CloudTasksUtils cloudTasksUtils,
      LockHandler lockHandler,
      String bucket,
      Duration lockTimeout,
      byte[] stagingKeyBytes,
      ValidationMode validationMode,
      GcsUtils gcsUtils) {
-    this.taskQueueUtils = taskQueueUtils;
+    this.cloudTasksUtils = cloudTasksUtils;
    this.lockHandler = lockHandler;
    this.bucket = bucket;
    this.lockTimeout = lockTimeout;
@@ -226,23 +226,35 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag
              logger.atInfo().log(
                  "Rolled forward %s on %s cursor to %s.", key.cursor(), tld, newPosition);
              RdeRevision.saveRevision(tld, watermark, mode, revision);
+              // Enqueueing a task is a side effect that is not undone if the transaction rolls
+              // back. So this may result in multiple copies of the same task being processed. This
+              // is fine because the RdeUploadAction is guarded by a lock and tracks progress by
+              // cursor. The BrdaCopyAction writes a file to GCS, which is an atomic action.
              if (mode == RdeMode.FULL) {
-                taskQueueUtils.enqueue(
-                    getQueue("rde-upload"),
-                    withUrl(RdeUploadAction.PATH).param(RequestParameters.PARAM_TLD, tld));
+                cloudTasksUtils.enqueue(
+                    "rde-upload",
+                    CloudTasksUtils.createPostTask(
+                        RdeUploadAction.PATH,
+                        Service.BACKEND.toString(),
+                        ImmutableMultimap.of(RequestParameters.PARAM_TLD, tld)));
              } else {
-                taskQueueUtils.enqueue(
-                    getQueue("brda"),
-                    withUrl(BrdaCopyAction.PATH)
-                        .param(RequestParameters.PARAM_TLD, tld)
-                        .param(RdeModule.PARAM_WATERMARK, watermark.toString()));
+                cloudTasksUtils.enqueue(
+                    "brda",
+                    CloudTasksUtils.createPostTask(
+                        BrdaCopyAction.PATH,
+                        Service.BACKEND.toString(),
+                        ImmutableMultimap.of(
+                            RequestParameters.PARAM_TLD,
+                            tld,
+                            RdeModule.PARAM_WATERMARK,
+                            watermark.toString())));
              }
            });
  }

  /** Injectible factory for creating {@link RdeStagingReducer}. */
  static class Factory {
-    @Inject TaskQueueUtils taskQueueUtils;
+    @Inject CloudTasksUtils cloudTasksUtils;
    @Inject LockHandler lockHandler;
    @Inject @Config("rdeBucket") String bucket;
    @Inject @Config("rdeStagingLockTimeout") Duration lockTimeout;
@@ -252,7 +264,7 @@ public final class RdeStagingReducer extends Reducer<PendingDeposit, DepositFrag

    RdeStagingReducer create(ValidationMode validationMode, GcsUtils gcsUtils) {
      return new RdeStagingReducer(
-          taskQueueUtils,
+          cloudTasksUtils,
          lockHandler,
          bucket,
          lockTimeout,
@@ -0,0 +1,58 @@
+// Copyright 2022 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.tools;
+
+import java.util.Optional;
+
+/**
+ * Enumerates the DNSSEC digest types for use with Delegation Signer records.
+ *
+ * <p>This also enforces the set of types that are valid for use with Cloud DNS. Customers cannot
+ * create DS records containing any other digest type.
+ *
+ * <p>The complete list can be found here:
+ * https://www.iana.org/assignments/ds-rr-types/ds-rr-types.xhtml
+ */
+public enum DigestType {
+  SHA1(1),
+  SHA256(2),
+  // Algorithm number 3 is GOST R 34.11-94 and is deliberately NOT SUPPORTED.
+  // This algorithm was reviewed by ise-crypto and deemed academically broken (b/207029800).
+  // In addition, RFC 8624 specifies that this algorithm MUST NOT be used for DNSSEC delegations.
+  // TODO(sarhabot@): Add note in Cloud DNS code to notify the Registry of any new changes to
+  // supported digest types.
+  SHA384(4);
+
+  private final int wireValue;
+
+  DigestType(int wireValue) {
+    this.wireValue = wireValue;
+  }
+
+  /** Fetches a DigestType enumeration constant by its IANA assigned value. */
+  public static Optional<DigestType> fromWireValue(int wireValue) {
+    for (DigestType alg : DigestType.values()) {
+      if (alg.getWireValue() == wireValue) {
+        return Optional.of(alg);
+      }
+    }
+    return Optional.empty();
+  }
+
+  /** Fetches a value in the range [0, 255] that encodes this DS digest type on the wire. */
+  public int getWireValue() {
+    return wireValue;
+  }
+}
@@ -16,6 +16,7 @@ package google.registry.tools;

 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableList.toImmutableList;
+import static google.registry.util.PreconditionsUtils.checkArgumentPresent;

 import com.beust.jcommander.IStringConverter;
 import com.google.auto.value.AutoValue;
@@ -25,6 +26,7 @@ import com.google.common.base.Splitter;
 import com.google.common.io.BaseEncoding;
 import com.google.template.soy.data.SoyListData;
 import com.google.template.soy.data.SoyMapData;
+import google.registry.flows.domain.DomainFlowUtils;
 import java.util.List;

@AutoValue
@@ -46,6 +48,15 @@ abstract class DsRecord {
        "digest should be even-lengthed hex, but is %s (length %s)",
        digest,
        digest.length());
+    checkArgumentPresent(
+        DigestType.fromWireValue(digestType),
+        String.format("DS record uses an unrecognized digest type: %d", digestType));
+
+    if (!DomainFlowUtils.validateAlgorithm(alg)) {
+      throw new IllegalArgumentException(
+          String.format("DS record uses an unrecognized algorithm: %d", alg));
+    }
+
    return new AutoValue_DsRecord(keyTag, alg, digestType, digest);
  }

@@ -106,6 +106,7 @@ import google.registry.flows.domain.DomainFlowUtils.FeeDescriptionParseException
 import google.registry.flows.domain.DomainFlowUtils.FeesMismatchException;
 import google.registry.flows.domain.DomainFlowUtils.FeesRequiredDuringEarlyAccessProgramException;
 import google.registry.flows.domain.DomainFlowUtils.FeesRequiredForPremiumNameException;
+import google.registry.flows.domain.DomainFlowUtils.InvalidDsRecordException;
 import google.registry.flows.domain.DomainFlowUtils.InvalidIdnDomainLabelException;
 import google.registry.flows.domain.DomainFlowUtils.InvalidPunycodeException;
 import google.registry.flows.domain.DomainFlowUtils.InvalidTcnIdChecksumException;
@@ -1002,6 +1003,22 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
    assertAboutEppExceptions().that(thrown).marshalsToXml();
  }

+  @TestOfyAndSql
+  void testFailure_secDnsInvalidDigestType() throws Exception {
+    setEppInput("domain_create_dsdata_bad_digest_types.xml");
+    persistContactsAndHosts();
+    EppException thrown = assertThrows(InvalidDsRecordException.class, this::runFlow);
+    assertAboutEppExceptions().that(thrown).marshalsToXml();
+  }
+
+  @TestOfyAndSql
+  void testFailure_secDnsInvalidAlgorithm() throws Exception {
+    setEppInput("domain_create_dsdata_bad_algorithms.xml");
+    persistContactsAndHosts();
+    EppException thrown = assertThrows(InvalidDsRecordException.class, this::runFlow);
+    assertAboutEppExceptions().that(thrown).marshalsToXml();
+  }
+
  @TestOfyAndSql
  void testFailure_wrongExtension() {
    setEppInput("domain_create_wrong_extension.xml");
@@ -72,6 +72,7 @@ import google.registry.flows.domain.DomainFlowUtils.DuplicateContactForRoleExcep
 import google.registry.flows.domain.DomainFlowUtils.EmptySecDnsUpdateException;
 import google.registry.flows.domain.DomainFlowUtils.FeesMismatchException;
 import google.registry.flows.domain.DomainFlowUtils.FeesRequiredForNonFreeOperationException;
+import google.registry.flows.domain.DomainFlowUtils.InvalidDsRecordException;
 import google.registry.flows.domain.DomainFlowUtils.LinkedResourceInPendingDeleteProhibitsOperationException;
 import google.registry.flows.domain.DomainFlowUtils.LinkedResourcesDoNotExistException;
 import google.registry.flows.domain.DomainFlowUtils.MaxSigLifeChangeNotSupportedException;
@@ -122,7 +123,7 @@ import org.junit.jupiter.api.extension.RegisterExtension;
 class DomainUpdateFlowTest extends ResourceFlowTestCase<DomainUpdateFlow, DomainBase> {

  private static final DelegationSignerData SOME_DSDATA =
-      DelegationSignerData.create(1, 2, 3, base16().decode("0123"));
+      DelegationSignerData.create(1, 2, 2, base16().decode("0123"));
  private static final ImmutableMap<String, String> OTHER_DSDATA_TEMPLATE_MAP =
      ImmutableMap.of(
          "KEY_TAG", "12346",
@@ -536,7 +537,7 @@ class DomainUpdateFlowTest extends ResourceFlowTestCase<DomainUpdateFlow, Domain
        "domain_update_dsdata_add.xml",
        ImmutableSet.of(SOME_DSDATA),
        ImmutableSet.of(SOME_DSDATA),
-        ImmutableMap.of("KEY_TAG", "1", "ALG", "2", "DIGEST_TYPE", "3", "DIGEST", "0123"));
+        ImmutableMap.of("KEY_TAG", "1", "ALG", "2", "DIGEST_TYPE", "2", "DIGEST", "0123"));
  }

  @TestOfyAndSql
@@ -556,8 +557,8 @@ class DomainUpdateFlowTest extends ResourceFlowTestCase<DomainUpdateFlow, Domain
        "domain_update_dsdata_add.xml",
        ImmutableSet.of(SOME_DSDATA),
        ImmutableSet.of(
-            SOME_DSDATA, DelegationSignerData.create(12346, 2, 3, base16().decode("0123"))),
-        ImmutableMap.of("KEY_TAG", "12346", "ALG", "2", "DIGEST_TYPE", "3", "DIGEST", "0123"));
+            SOME_DSDATA, DelegationSignerData.create(12346, 2, 2, base16().decode("0123"))),
+        ImmutableMap.of("KEY_TAG", "12346", "ALG", "2", "DIGEST_TYPE", "2", "DIGEST", "0123"));
  }

  @TestOfyAndSql
@@ -565,8 +566,8 @@ class DomainUpdateFlowTest extends ResourceFlowTestCase<DomainUpdateFlow, Domain
    doSecDnsSuccessfulTest(
        "domain_update_dsdata_add.xml",
        ImmutableSet.of(SOME_DSDATA),
-        ImmutableSet.of(SOME_DSDATA, DelegationSignerData.create(1, 8, 3, base16().decode("0123"))),
-        ImmutableMap.of("KEY_TAG", "1", "ALG", "8", "DIGEST_TYPE", "3", "DIGEST", "0123"));
+        ImmutableSet.of(SOME_DSDATA, DelegationSignerData.create(1, 8, 2, base16().decode("0123"))),
+        ImmutableMap.of("KEY_TAG", "1", "ALG", "8", "DIGEST_TYPE", "2", "DIGEST", "0123"));
  }

  @TestOfyAndSql
@@ -583,15 +584,15 @@ class DomainUpdateFlowTest extends ResourceFlowTestCase<DomainUpdateFlow, Domain
    doSecDnsSuccessfulTest(
        "domain_update_dsdata_add.xml",
        ImmutableSet.of(SOME_DSDATA),
-        ImmutableSet.of(SOME_DSDATA, DelegationSignerData.create(1, 2, 3, base16().decode("4567"))),
-        ImmutableMap.of("KEY_TAG", "1", "ALG", "2", "DIGEST_TYPE", "3", "DIGEST", "4567"));
+        ImmutableSet.of(SOME_DSDATA, DelegationSignerData.create(1, 2, 2, base16().decode("4567"))),
+        ImmutableMap.of("KEY_TAG", "1", "ALG", "2", "DIGEST_TYPE", "2", "DIGEST", "4567"));
  }

  @TestOfyAndSql
  void testSuccess_secDnsAddToMaxRecords() throws Exception {
    ImmutableSet.Builder<DelegationSignerData> builder = new ImmutableSet.Builder<>();
    for (int i = 0; i < 7; ++i) {
-      builder.add(DelegationSignerData.create(i, 2, 3, new byte[] {0, 1, 2}));
+      builder.add(DelegationSignerData.create(i, 2, 2, new byte[] {0, 1, 2}));
    }
    ImmutableSet<DelegationSignerData> commonDsData = builder.build();

@@ -643,7 +644,7 @@ class DomainUpdateFlowTest extends ResourceFlowTestCase<DomainUpdateFlow, Domain
  void testSuccess_secDnsAddRemoveToMaxRecords() throws Exception {
    ImmutableSet.Builder<DelegationSignerData> builder = new ImmutableSet.Builder<>();
    for (int i = 0; i < 7; ++i) {
-      builder.add(DelegationSignerData.create(i, 2, 3, new byte[] {0, 1, 2}));
+      builder.add(DelegationSignerData.create(i, 2, 2, new byte[] {0, 1, 2}));
    }
    ImmutableSet<DelegationSignerData> commonDsData = builder.build();

@@ -813,11 +814,69 @@ class DomainUpdateFlowTest extends ResourceFlowTestCase<DomainUpdateFlow, Domain
        MaxSigLifeChangeNotSupportedException.class, "domain_update_maxsiglife.xml");
  }

+  @TestOfyAndSql
+  void testFailure_secDnsInvalidDigestType() throws Exception {
+    setEppInput("domain_update_dsdata_add.xml", OTHER_DSDATA_TEMPLATE_MAP);
+    persistResource(
+        newDomainBase(getUniqueIdFromCommand())
+            .asBuilder()
+            .setDsData(ImmutableSet.of(DelegationSignerData.create(1, 2, 3, new byte[] {0, 1, 2})))
+            .build());
+    EppException thrown = assertThrows(InvalidDsRecordException.class, this::runFlow);
+    assertAboutEppExceptions().that(thrown).marshalsToXml();
+  }
+
+  @TestOfyAndSql
+  void testFailure_secDnsMultipleInvalidDigestTypes() throws Exception {
+    setEppInput("domain_update_dsdata_add.xml", OTHER_DSDATA_TEMPLATE_MAP);
+    persistResource(
+        newDomainBase(getUniqueIdFromCommand())
+            .asBuilder()
+            .setDsData(
+                ImmutableSet.of(
+                    DelegationSignerData.create(1, 2, 3, new byte[] {0, 1, 2}),
+                    DelegationSignerData.create(2, 2, 6, new byte[] {0, 1, 2})))
+            .build());
+    EppException thrown = assertThrows(InvalidDsRecordException.class, this::runFlow);
+    assertThat(thrown).hasMessageThat().contains("digestType=3");
+    assertThat(thrown).hasMessageThat().contains("digestType=6");
+    assertAboutEppExceptions().that(thrown).marshalsToXml();
+  }
+
+  @TestOfyAndSql
+  void testFailure_secDnsInvalidAlgorithm() throws Exception {
+    setEppInput("domain_update_dsdata_add.xml", OTHER_DSDATA_TEMPLATE_MAP);
+    persistResource(
+        newDomainBase(getUniqueIdFromCommand())
+            .asBuilder()
+            .setDsData(ImmutableSet.of(DelegationSignerData.create(1, 99, 2, new byte[] {0, 1, 2})))
+            .build());
+    EppException thrown = assertThrows(InvalidDsRecordException.class, this::runFlow);
+    assertAboutEppExceptions().that(thrown).marshalsToXml();
+  }
+
+  @TestOfyAndSql
+  void testFailure_secDnsMultipleInvalidAlgorithms() throws Exception {
+    setEppInput("domain_update_dsdata_add.xml", OTHER_DSDATA_TEMPLATE_MAP);
+    persistResource(
+        newDomainBase(getUniqueIdFromCommand())
+            .asBuilder()
+            .setDsData(
+                ImmutableSet.of(
+                    DelegationSignerData.create(1, 998, 2, new byte[] {0, 1, 2}),
+                    DelegationSignerData.create(2, 99, 2, new byte[] {0, 1, 2})))
+            .build());
+    EppException thrown = assertThrows(InvalidDsRecordException.class, this::runFlow);
+    assertThat(thrown).hasMessageThat().contains("algorithm=998");
+    assertThat(thrown).hasMessageThat().contains("algorithm=99");
+    assertAboutEppExceptions().that(thrown).marshalsToXml();
+  }
+
  @TestOfyAndSql
  void testFailure_secDnsTooManyDsRecords() throws Exception {
    ImmutableSet.Builder<DelegationSignerData> builder = new ImmutableSet.Builder<>();
    for (int i = 0; i < 8; ++i) {
-      builder.add(DelegationSignerData.create(i, 2, 3, new byte[] {0, 1, 2}));
+      builder.add(DelegationSignerData.create(i, 2, 2, new byte[] {0, 1, 2}));
    }

    setEppInput("domain_update_dsdata_add.xml", OTHER_DSDATA_TEMPLATE_MAP);
@@ -16,6 +16,7 @@ package google.registry.persistence.transaction;

 import static com.google.common.truth.Truth.assertThat;
 import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
+import static google.registry.persistence.transaction.TransactionManagerFactory.replicaJpaTm;
 import static google.registry.testing.DatabaseHelper.insertInDb;
 import static org.junit.jupiter.api.Assertions.assertThrows;

@@ -62,6 +63,17 @@ public class JpaTransactionManagerExtensionTest {
            });
  }

+  @Test
+  void testReplicaJpaTm() {
+    TestEntity testEntity = new TestEntity("foo", "bar");
+    assertThat(
+            assertThrows(
+                PersistenceException.class,
+                () -> replicaJpaTm().transact(() -> replicaJpaTm().put(testEntity))))
+        .hasMessageThat()
+        .isEqualTo("Error while committing the transaction");
+  }
+
  @Test
  void testExtraParameters() {
    // This test verifies that 1) withEntityClass() has registered TestEntity and 2) The table
@@ -15,7 +15,6 @@
 package google.registry.rdap;

 import static com.google.common.truth.Truth.assertThat;
-import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 import static google.registry.rdap.RdapTestHelper.assertThat;
 import static google.registry.rdap.RdapTestHelper.parseJsonObject;
 import static google.registry.request.Action.Method.POST;
@@ -46,7 +45,6 @@ import google.registry.model.registrar.Registrar;
 import google.registry.model.reporting.HistoryEntry;
 import google.registry.model.tld.Registry;
 import google.registry.persistence.VKey;
-import google.registry.persistence.transaction.ReplicaSimulatingJpaTransactionManager;
 import google.registry.rdap.RdapMetrics.EndpointType;
 import google.registry.rdap.RdapMetrics.SearchType;
 import google.registry.rdap.RdapMetrics.WildcardType;
@@ -376,7 +374,6 @@ class RdapDomainSearchActionTest extends RdapSearchActionTestCase<RdapDomainSear
    action.nsIpParam = Optional.empty();
    action.cursorTokenParam = Optional.empty();
    action.requestPath = actionPath;
-    action.readOnlyJpaTm = jpaTm();
  }

  private JsonObject generateExpectedJsonForTwoDomainsNsReply() {
@@ -724,18 +721,6 @@ class RdapDomainSearchActionTest extends RdapSearchActionTestCase<RdapDomainSear
    verifyMetrics(SearchType.BY_DOMAIN_NAME, Optional.of(1L));
  }

-  @TestSqlOnly
-  void testDomainMatch_readOnlyReplica() {
-    login("evilregistrar");
-    rememberWildcardType("cat.lol");
-    action.readOnlyJpaTm = new ReplicaSimulatingJpaTransactionManager(jpaTm());
-    action.nameParam = Optional.of("cat.lol");
-    action.parameterMap = ImmutableListMultimap.of("name", "cat.lol");
-    action.run();
-    assertThat(response.getPayload()).contains("Yes Virginia <script>");
-    assertThat(response.getStatus()).isEqualTo(200);
-  }
-
  @TestOfyAndSql
  void testDomainMatch_foundWithUpperCase() {
    login("evilregistrar");
@@ -16,6 +16,7 @@ package google.registry.rde;

 import static com.google.common.truth.Truth.assertThat;
 import static com.google.common.truth.Truth.assertWithMessage;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.testing.SystemInfo.hasCommand;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.jupiter.api.Assumptions.assumeTrue;
@@ -27,6 +28,8 @@ import com.google.common.io.CharStreams;
 import com.google.common.io.Files;
 import google.registry.gcs.GcsUtils;
 import google.registry.keyring.api.Keyring;
+import google.registry.model.rde.RdeMode;
+import google.registry.model.rde.RdeRevision;
 import google.registry.testing.AppEngineExtension;
 import google.registry.testing.BouncyCastleProviderExtension;
 import google.registry.testing.FakeKeyringModule;
@@ -109,6 +112,10 @@ public class BrdaCopyActionTest {
    action.receiverKey = receiverKey;
    action.signingKey = signingKey;
    action.stagingDecryptionKey = decryptKey;
+    tm().transact(
+            () -> {
+              RdeRevision.saveRevision("lol", DateTime.parse("2010-10-17TZ"), RdeMode.THIN, 0);
+            });
  }

  @ParameterizedTest
@@ -27,7 +27,6 @@ import static google.registry.testing.DatabaseHelper.persistResource;
 import static google.registry.testing.DatabaseHelper.persistResourceWithCommitLog;
 import static google.registry.testing.TaskQueueHelper.assertAtLeastOneTaskIsEnqueued;
 import static google.registry.testing.TaskQueueHelper.assertNoTasksEnqueued;
-import static google.registry.testing.TaskQueueHelper.assertTasksEnqueued;
 import static google.registry.testing.TestDataHelper.loadFile;
 import static google.registry.tldconfig.idn.IdnTableEnum.EXTENDED_LATIN;
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -50,17 +49,15 @@ import google.registry.model.ofy.Ofy;
 import google.registry.model.tld.Registry;
 import google.registry.request.HttpException.BadRequestException;
 import google.registry.request.RequestParameters;
+import google.registry.testing.CloudTasksHelper;
+import google.registry.testing.CloudTasksHelper.TaskMatcher;
 import google.registry.testing.FakeClock;
 import google.registry.testing.FakeKeyringModule;
 import google.registry.testing.FakeLockHandler;
 import google.registry.testing.FakeResponse;
 import google.registry.testing.InjectExtension;
-import google.registry.testing.TaskQueueHelper.TaskMatcher;
 import google.registry.testing.mapreduce.MapreduceTestCase;
 import google.registry.tldconfig.idn.IdnTableEnum;
-import google.registry.util.Retrier;
-import google.registry.util.SystemSleeper;
-import google.registry.util.TaskQueueUtils;
 import google.registry.xjc.XjcXmlTransformer;
 import google.registry.xjc.rde.XjcRdeContentType;
 import google.registry.xjc.rde.XjcRdeDeposit;
@@ -100,10 +97,19 @@ public class RdeStagingActionDatastoreTest extends MapreduceTestCase<RdeStagingA

  @RegisterExtension public final InjectExtension inject = new InjectExtension();

-  private final FakeClock clock = new FakeClock();
+  /**
+   * Without autoIncrement mode, the fake clock won't advance between Mapper and Reducer
+   * transactions when action is invoked, resulting in rolled back reducer transaction (due to
+   * TimestampInversionException if both transactions are mapped to the same CommitLog bucket) and
+   * multiple RdeUplaod/BrdaCopy tasks being enqueued (due to transaction retries, since Cloud Tasks
+   * enqueuing is not transactional with Datastore transactions).
+   */
+  private final FakeClock clock = new FakeClock().setAutoIncrementByOneMilli();
+
  private final FakeResponse response = new FakeResponse();
  private final GcsUtils gcsUtils = new GcsUtils(LocalStorageHelper.getOptions());
  private final List<? super XjcRdeContentType> alreadyExtracted = new ArrayList<>();
+  private final CloudTasksHelper cloudTasksHelper = new CloudTasksHelper();

  private static PGPPublicKey encryptKey;
  private static PGPPrivateKey decryptKey;
@@ -124,7 +130,7 @@ public class RdeStagingActionDatastoreTest extends MapreduceTestCase<RdeStagingA
    action.mrRunner = makeDefaultRunner();
    action.lenient = false;
    action.reducerFactory = new RdeStagingReducer.Factory();
-    action.reducerFactory.taskQueueUtils = new TaskQueueUtils(new Retrier(new SystemSleeper(), 1));
+    action.reducerFactory.cloudTasksUtils = cloudTasksHelper.getTestCloudTasksUtils();
    action.reducerFactory.lockHandler = new FakeLockHandler(true);
    action.reducerFactory.bucket = "rde-bucket";
    action.reducerFactory.lockTimeout = Duration.standardHours(1);
@@ -467,11 +473,13 @@ public class RdeStagingActionDatastoreTest extends MapreduceTestCase<RdeStagingA
    clock.setTo(DateTime.parse("2000-01-04TZ")); // Tuesday
    action.run();
    executeTasksUntilEmpty("mapreduce", clock);
-    assertTasksEnqueued("rde-upload",
-        new TaskMatcher()
-            .url(RdeUploadAction.PATH)
-            .param(RequestParameters.PARAM_TLD, "lol"));
-    assertTasksEnqueued("brda",
+    // TODO(b/217773051): duplicate tasks are possible though unlikely. Consider if below calls are
+    // appropriate since they don't allow duplicates.
+    cloudTasksHelper.assertTasksEnqueued(
+        "rde-upload",
+        new TaskMatcher().url(RdeUploadAction.PATH).param(RequestParameters.PARAM_TLD, "lol"));
+    cloudTasksHelper.assertTasksEnqueued(
+        "brda",
        new TaskMatcher()
            .url(BrdaCopyAction.PATH)
            .param(RequestParameters.PARAM_TLD, "lol")
@@ -20,8 +20,6 @@ import static google.registry.model.rde.RdeMode.FULL;
 import static google.registry.model.rde.RdeMode.THIN;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.testing.DatabaseHelper.createTld;
-import static google.registry.testing.TaskQueueHelper.assertNoTasksEnqueued;
-import static google.registry.testing.TaskQueueHelper.assertTasksEnqueued;
 import static google.registry.util.ResourceUtils.readResourceUtf8;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -41,13 +39,10 @@ import google.registry.model.rde.RdeRevision;
 import google.registry.model.tld.Registry;
 import google.registry.request.RequestParameters;
 import google.registry.testing.AppEngineExtension;
-import google.registry.testing.FakeClock;
+import google.registry.testing.CloudTasksHelper;
+import google.registry.testing.CloudTasksHelper.TaskMatcher;
 import google.registry.testing.FakeKeyringModule;
 import google.registry.testing.FakeLockHandler;
-import google.registry.testing.FakeSleeper;
-import google.registry.testing.TaskQueueHelper.TaskMatcher;
-import google.registry.util.Retrier;
-import google.registry.util.TaskQueueUtils;
 import google.registry.xml.ValidationMode;
 import java.io.IOException;
 import java.util.Iterator;
@@ -74,6 +69,7 @@ class RdeStagingReducerTest {
  private static final PGPPublicKey encryptionKey =
      new FakeKeyringModule().get().getRdeStagingEncryptionKey();
  private static final DateTime now = DateTime.parse("2000-01-01TZ");
+  private final CloudTasksHelper cloudTasksHelper = new CloudTasksHelper();

  private Fragments brdaFragments =
      new Fragments(
@@ -94,7 +90,7 @@ class RdeStagingReducerTest {

  private RdeStagingReducer reducer =
      new RdeStagingReducer(
-          new TaskQueueUtils(new Retrier(new FakeSleeper(new FakeClock()), 1)),
+          cloudTasksHelper.getTestCloudTasksUtils(),
          new FakeLockHandler(true),
          GCS_BUCKET,
          Duration.ZERO,
@@ -133,7 +129,7 @@ class RdeStagingReducerTest {
    assertThat(loadCursorTime(CursorType.BRDA))
        .isEquivalentAccordingToCompareTo(now.plus(Duration.standardDays(1)));
    assertThat(loadRevision(THIN)).isEqualTo(1);
-    assertTasksEnqueued(
+    cloudTasksHelper.assertTasksEnqueued(
        "brda",
        new TaskMatcher()
            .url(BrdaCopyAction.PATH)
@@ -159,7 +155,7 @@ class RdeStagingReducerTest {
    // No extra operations in manual mode.
    assertThat(loadCursorTime(CursorType.BRDA)).isEquivalentAccordingToCompareTo(now);
    assertThat(loadRevision(THIN)).isEqualTo(0);
-    assertNoTasksEnqueued("brda");
+    cloudTasksHelper.assertNoTasksEnqueued("brda");
  }

  @Test
@@ -179,7 +175,7 @@ class RdeStagingReducerTest {
    assertThat(loadCursorTime(CursorType.RDE_STAGING))
        .isEquivalentAccordingToCompareTo(now.plus(Duration.standardDays(1)));
    assertThat(loadRevision(FULL)).isEqualTo(1);
-    assertTasksEnqueued(
+    cloudTasksHelper.assertTasksEnqueued(
        "rde-upload",
        new TaskMatcher().url(RdeUploadAction.PATH).param(RequestParameters.PARAM_TLD, "soy"));
  }
@@ -200,7 +196,7 @@ class RdeStagingReducerTest {
    // No extra operations in manual mode.
    assertThat(loadCursorTime(CursorType.RDE_STAGING)).isEquivalentAccordingToCompareTo(now);
    assertThat(loadRevision(FULL)).isEqualTo(0);
-    assertNoTasksEnqueued("rde-upload");
+    cloudTasksHelper.assertNoTasksEnqueued("rde-upload");
  }

  private static void compareLength(String outputFile, String lengthFilename) throws IOException {
@@ -39,6 +39,7 @@ import com.google.common.collect.Multimaps;
 import com.google.common.net.HttpHeaders;
 import com.google.common.net.MediaType;
 import com.google.common.truth.Truth8;
+import com.google.protobuf.Timestamp;
 import google.registry.model.ImmutableObject;
 import google.registry.util.CloudTasksUtils;
 import google.registry.util.Retrier;
@@ -195,6 +196,7 @@ public class CloudTasksHelper implements Serializable {
    // tests.
    HttpMethod method = HttpMethod.POST;
    String url;
+    Timestamp scheduleTime;
    Multimap<String, String> headers = ArrayListMultimap.create();
    Multimap<String, String> params = ArrayListMultimap.create();

@@ -216,6 +218,7 @@ public class CloudTasksHelper implements Serializable {
          Ascii.toLowerCase(task.getAppEngineHttpRequest().getAppEngineRouting().getService());
      method = task.getAppEngineHttpRequest().getHttpMethod();
      url = uri.getPath();
+      scheduleTime = task.getScheduleTime();
      ImmutableMultimap.Builder<String, String> headerBuilder = new ImmutableMultimap.Builder<>();
      task.getAppEngineHttpRequest()
          .getHeadersMap()
@@ -251,6 +254,7 @@ public class CloudTasksHelper implements Serializable {
      builder.put("url", url);
      builder.put("headers", headers);
      builder.put("params", params);
+      builder.put("scheduleTime", scheduleTime);
      return Maps.filterValues(builder, not(in(asList(null, "", Collections.EMPTY_MAP))));
    }
  }
@@ -293,6 +297,11 @@ public class CloudTasksHelper implements Serializable {
      return this;
    }

+    public TaskMatcher scheduleTime(Timestamp scheduleTime) {
+      expected.scheduleTime = scheduleTime;
+      return this;
+    }
+
    public TaskMatcher param(String key, String value) {
      checkNotNull(value, "Test error: A param can never have a null value, so don't assert it");
      expected.params.put(key, value);
@@ -316,6 +325,8 @@ public class CloudTasksHelper implements Serializable {
     *
     * <p>Match fails if any headers or params expected on the TaskMatcher are not found on the
     * Task. Note that the inverse is not true (i.e. there may be extra headers on the Task).
+     *
+     * <p>Schedule time by default is Timestamp.getDefaultInstance() or null.
     */
    @Override
    public boolean test(@Nonnull Task task) {
@@ -324,6 +335,8 @@ public class CloudTasksHelper implements Serializable {
          && (expected.url == null || Objects.equals(expected.url, actual.url))
          && (expected.method == null || Objects.equals(expected.method, actual.method))
          && (expected.service == null || Objects.equals(expected.service, actual.service))
+          && (expected.scheduleTime == null
+              || Objects.equals(expected.scheduleTime, actual.scheduleTime))
          && containsEntries(actual.params, expected.params)
          && containsEntries(actual.headers, expected.headers);
    }
@@ -50,7 +50,7 @@ class CreateDomainCommandTest extends EppToolCommandTestCase<CreateDomainCommand
        "--admins=crr-admin",
        "--techs=crr-tech",
        "--password=2fooBAR",
-        "--ds_records=1 2 3 abcd,4 5 6 EF01",
+        "--ds_records=1 2 2 abcd,4 5 1 EF01",
        "--ds_records=60485 5  2  D4B7D520E7BB5F0F67674A0CCEB1E3E0614B93C4F9E99B8383F6A1E4469DA50A",
        "example.tld");
    eppVerifier.verifySent("domain_create_complete.xml");
@@ -66,7 +66,7 @@ class CreateDomainCommandTest extends EppToolCommandTestCase<CreateDomainCommand
        "--admins=crr-admin",
        "--techs=crr-tech",
        "--password=2fooBAR",
-        "--ds_records=1 2 3 abcd,4 5 6 EF01",
+        "--ds_records=1 2 2 abcd,4 5 1 EF01",
        "--ds_records=60485 5  2  D4B7D520E7BB5F0F67674A0CCEB1E3E0614B93C4F9E99B8383F6A1E4469DA50A",
        "example.tld");
    eppVerifier.verifySent("domain_create_complete.xml");
@@ -314,6 +314,38 @@ class CreateDomainCommandTest extends EppToolCommandTestCase<CreateDomainCommand
    assertThat(thrown).hasMessageThat().contains("--period");
  }

+  @Test
+  void testFailure_invalidDigestType() {
+    IllegalArgumentException thrown =
+        assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                runCommandForced(
+                    "--client=NewRegistrar",
+                    "--registrant=crr-admin",
+                    "--admins=crr-admin",
+                    "--techs=crr-tech",
+                    "--ds_records=1 2 3 abcd",
+                    "example.tld"));
+    assertThat(thrown).hasMessageThat().isEqualTo("DS record uses an unrecognized digest type: 3");
+  }
+
+  @Test
+  void testFailure_invalidAlgorithm() {
+    IllegalArgumentException thrown =
+        assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                runCommandForced(
+                    "--client=NewRegistrar",
+                    "--registrant=crr-admin",
+                    "--admins=crr-admin",
+                    "--techs=crr-tech",
+                    "--ds_records=1 999 4 abcd",
+                    "example.tld"));
+    assertThat(thrown).hasMessageThat().isEqualTo("DS record uses an unrecognized algorithm: 999");
+  }
+
  @Test
  void testFailure_dsRecordsNot4Parts() {
    IllegalArgumentException thrown =
@@ -86,12 +86,12 @@ class UpdateDomainCommandTest extends EppToolCommandTestCase<UpdateDomainCommand
        "--add_admins=crr-admin2",
        "--add_techs=crr-tech2",
        "--add_statuses=serverDeleteProhibited",
-        "--add_ds_records=1 2 3 abcd,4 5 6 EF01",
+        "--add_ds_records=1 2 2 abcd,4 5 1 EF01",
        "--remove_nameservers=ns3.zdns.google,ns4.zdns.google",
        "--remove_admins=crr-admin1",
        "--remove_techs=crr-tech1",
        "--remove_statuses=serverHold",
-        "--remove_ds_records=7 8 9 12ab,6 5 4 34CD",
+        "--remove_ds_records=7 8 1 12ab,6 5 4 34CD",
        "--registrant=crr-admin",
        "--password=2fooBAR",
        "example.tld");
@@ -106,12 +106,12 @@ class UpdateDomainCommandTest extends EppToolCommandTestCase<UpdateDomainCommand
        "--add_admins=crr-admin2",
        "--add_techs=crr-tech2",
        "--add_statuses=serverDeleteProhibited",
-        "--add_ds_records=1 2 3 abcd,4 5 6 EF01",
+        "--add_ds_records=1 2 2 abcd,4 5 1 EF01",
        "--remove_nameservers=ns[3-4].zdns.google",
        "--remove_admins=crr-admin1",
        "--remove_techs=crr-tech1",
        "--remove_statuses=serverHold",
-        "--remove_ds_records=7 8 9 12ab,6 5 4 34CD",
+        "--remove_ds_records=7 8 1 12ab,6 5 4 34CD",
        "--registrant=crr-admin",
        "--password=2fooBAR",
        "example.tld");
@@ -128,12 +128,12 @@ class UpdateDomainCommandTest extends EppToolCommandTestCase<UpdateDomainCommand
        "--add_admins=crr-admin2",
        "--add_techs=crr-tech2",
        "--add_statuses=serverDeleteProhibited",
-        "--add_ds_records=1 2 3 abcd,4 5 6 EF01",
+        "--add_ds_records=1 2 2 abcd,4 5 1 EF01",
        "--remove_nameservers=ns[3-4].zdns.google",
        "--remove_admins=crr-admin1",
        "--remove_techs=crr-tech1",
        "--remove_statuses=serverHold",
-        "--remove_ds_records=7 8 9 12ab,6 5 4 34CD",
+        "--remove_ds_records=7 8 1 12ab,6 5 4 34CD",
        "--registrant=crr-admin",
        "--password=2fooBAR",
        "example.tld",
@@ -186,7 +186,7 @@ class UpdateDomainCommandTest extends EppToolCommandTestCase<UpdateDomainCommand
        "--add_admins=crr-admin2",
        "--add_techs=crr-tech2",
        "--add_statuses=serverDeleteProhibited",
-        "--add_ds_records=1 2 3 abcd,4 5 6 EF01",
+        "--add_ds_records=1 2 2 abcd,4 5 1 EF01",
        "example.tld");
    eppVerifier.verifySent("domain_update_add.xml");
  }
@@ -199,7 +199,7 @@ class UpdateDomainCommandTest extends EppToolCommandTestCase<UpdateDomainCommand
        "--remove_admins=crr-admin1",
        "--remove_techs=crr-tech1",
        "--remove_statuses=serverHold",
-        "--remove_ds_records=7 8 9 12ab,6 5 4 34CD",
+        "--remove_ds_records=7 8 1 12ab,6 5 4 34CD",
        "example.tld");
    eppVerifier.verifySent("domain_update_remove.xml");
  }
@@ -277,8 +277,7 @@ class UpdateDomainCommandTest extends EppToolCommandTestCase<UpdateDomainCommand

  @TestOfyAndSql
  void testSuccess_setDsRecords() throws Exception {
-    runCommandForced(
-        "--client=NewRegistrar", "--ds_records=1 2 3 abcd,4 5 6 EF01", "example.tld");
+    runCommandForced("--client=NewRegistrar", "--ds_records=1 2 2 abcd,4 5 1 EF01", "example.tld");
    eppVerifier.verifySent("domain_update_set_ds_records.xml");
  }

@@ -286,7 +285,7 @@ class UpdateDomainCommandTest extends EppToolCommandTestCase<UpdateDomainCommand
  void testSuccess_setDsRecords_withUnneededClear() throws Exception {
    runCommandForced(
        "--client=NewRegistrar",
-        "--ds_records=1 2 3 abcd,4 5 6 EF01",
+        "--ds_records=1 2 2 abcd,4 5 1 EF01",
        "--clear_ds_records",
        "example.tld");
    eppVerifier.verifySent("domain_update_set_ds_records.xml");
@@ -630,6 +629,28 @@ class UpdateDomainCommandTest extends EppToolCommandTestCase<UpdateDomainCommand
                + "you cannot use the add_statuses and remove_statuses flags.");
  }

+  @TestOfyAndSql
+  void testFailure_invalidDsRecordAlgorithm() {
+    IllegalArgumentException thrown =
+        assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                runCommandForced(
+                    "--client=NewRegistrar", "--add_ds_records=1 299 2 abcd", "example.tld"));
+    assertThat(thrown).hasMessageThat().isEqualTo("DS record uses an unrecognized algorithm: 299");
+  }
+
+  @TestOfyAndSql
+  void testFailure_invalidDsRecordDigestType() {
+    IllegalArgumentException thrown =
+        assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                runCommandForced(
+                    "--client=NewRegistrar", "--add_ds_records=1 2 3 abcd", "example.tld"));
+    assertThat(thrown).hasMessageThat().isEqualTo("DS record uses an unrecognized digest type: 3");
+  }
+
  @TestOfyAndSql
  void testFailure_provideDsRecordsAndAddDsRecords() {
    IllegalArgumentException thrown =
@@ -638,8 +659,8 @@ class UpdateDomainCommandTest extends EppToolCommandTestCase<UpdateDomainCommand
            () ->
                runCommandForced(
                    "--client=NewRegistrar",
-                    "--add_ds_records=1 2 3 abcd",
-                    "--ds_records=4 5 6 EF01",
+                    "--add_ds_records=1 2 2 abcd",
+                    "--ds_records=4 5 1 EF01",
                    "example.tld"));
    assertThat(thrown)
        .hasMessageThat()
@@ -656,8 +677,8 @@ class UpdateDomainCommandTest extends EppToolCommandTestCase<UpdateDomainCommand
            () ->
                runCommandForced(
                    "--client=NewRegistrar",
-                    "--remove_ds_records=7 8 9 12ab",
-                    "--ds_records=4 5 6 EF01",
+                    "--remove_ds_records=7 8 1 12ab",
+                    "--ds_records=4 5 1 EF01",
                    "example.tld"));
    assertThat(thrown)
        .hasMessageThat()
@@ -674,7 +695,7 @@ class UpdateDomainCommandTest extends EppToolCommandTestCase<UpdateDomainCommand
            () ->
                runCommandForced(
                    "--client=NewRegistrar",
-                    "--add_ds_records=1 2 3 abcd",
+                    "--add_ds_records=1 2 2 abcd",
                    "--clear_ds_records",
                    "example.tld"));
    assertThat(thrown)
@@ -692,7 +713,7 @@ class UpdateDomainCommandTest extends EppToolCommandTestCase<UpdateDomainCommand
            () ->
                runCommandForced(
                    "--client=NewRegistrar",
-                    "--remove_ds_records=7 8 9 12ab",
+                    "--remove_ds_records=7 8 1 12ab",
                    "--clear_ds_records",
                    "example.tld"));
    assertThat(thrown)
@@ -0,0 +1,77 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<epp xmlns="urn:ietf:params:xml:ns:epp-1.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <command>
+    <create>
+      <domain:create
+          xmlns:domain="urn:ietf:params:xml:ns:domain-1.0">
+        <domain:name>example.tld</domain:name>
+        <domain:period unit="y">2</domain:period>
+        <domain:ns>
+          <domain:hostObj>ns1.example.net</domain:hostObj>
+          <domain:hostObj>ns2.example.net</domain:hostObj>
+        </domain:ns>
+        <domain:registrant>jd1234</domain:registrant>
+        <domain:contact type="admin">sh8013</domain:contact>
+        <domain:contact type="tech">sh8013</domain:contact>
+        <domain:authInfo>
+          <domain:pw>2fooBAR</domain:pw>
+        </domain:authInfo>
+      </domain:create>
+    </create>
+    <extension>
+      <secDNS:create
+          xmlns:secDNS="urn:ietf:params:xml:ns:secDNS-1.1">
+        <secDNS:dsData>
+          <secDNS:keyTag>12345</secDNS:keyTag>
+          <secDNS:alg>99</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12346</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12347</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12348</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12349</secDNS:keyTag>
+          <secDNS:alg>98</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12350</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12351</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12352</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+      </secDNS:create>
+    </extension>
+    <clTRID>ABC-12345</clTRID>
+  </command>
+</epp>
@@ -0,0 +1,77 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<epp xmlns="urn:ietf:params:xml:ns:epp-1.0"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+  <command>
+    <create>
+      <domain:create
+          xmlns:domain="urn:ietf:params:xml:ns:domain-1.0">
+        <domain:name>example.tld</domain:name>
+        <domain:period unit="y">2</domain:period>
+        <domain:ns>
+          <domain:hostObj>ns1.example.net</domain:hostObj>
+          <domain:hostObj>ns2.example.net</domain:hostObj>
+        </domain:ns>
+        <domain:registrant>jd1234</domain:registrant>
+        <domain:contact type="admin">sh8013</domain:contact>
+        <domain:contact type="tech">sh8013</domain:contact>
+        <domain:authInfo>
+          <domain:pw>2fooBAR</domain:pw>
+        </domain:authInfo>
+      </domain:create>
+    </create>
+    <extension>
+      <secDNS:create
+          xmlns:secDNS="urn:ietf:params:xml:ns:secDNS-1.1">
+        <secDNS:dsData>
+          <secDNS:keyTag>12345</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>100</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12346</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>3</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12347</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12348</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12349</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12350</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12351</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+        <secDNS:dsData>
+          <secDNS:keyTag>12352</secDNS:keyTag>
+          <secDNS:alg>3</secDNS:alg>
+          <secDNS:digestType>1</secDNS:digestType>
+          <secDNS:digest>49FD46E6C4B45C55D4AC</secDNS:digest>
+        </secDNS:dsData>
+      </secDNS:create>
+    </extension>
+    <clTRID>ABC-12345</clTRID>
+  </command>
+</epp>
@@ -25,13 +25,13 @@
        <secDNS:dsData>
          <secDNS:keyTag>1</secDNS:keyTag>
          <secDNS:alg>2</secDNS:alg>
-          <secDNS:digestType>3</secDNS:digestType>
+          <secDNS:digestType>2</secDNS:digestType>
          <secDNS:digest>ABCD</secDNS:digest>
        </secDNS:dsData>
        <secDNS:dsData>
          <secDNS:keyTag>4</secDNS:keyTag>
          <secDNS:alg>5</secDNS:alg>
-          <secDNS:digestType>6</secDNS:digestType>
+          <secDNS:digestType>1</secDNS:digestType>
          <secDNS:digest>EF01</secDNS:digest>
        </secDNS:dsData>
        <secDNS:dsData>
@@ -22,13 +22,13 @@
          <secDNS:dsData>
            <secDNS:keyTag>1</secDNS:keyTag>
            <secDNS:alg>2</secDNS:alg>
-            <secDNS:digestType>3</secDNS:digestType>
+            <secDNS:digestType>2</secDNS:digestType>
            <secDNS:digest>ABCD</secDNS:digest>
          </secDNS:dsData>
          <secDNS:dsData>
            <secDNS:keyTag>4</secDNS:keyTag>
            <secDNS:alg>5</secDNS:alg>
-            <secDNS:digestType>6</secDNS:digestType>
+            <secDNS:digestType>1</secDNS:digestType>
            <secDNS:digest>EF01</secDNS:digest>
          </secDNS:dsData>
        </secDNS:add>
@@ -37,7 +37,7 @@
          <secDNS:dsData>
            <secDNS:keyTag>7</secDNS:keyTag>
            <secDNS:alg>8</secDNS:alg>
-            <secDNS:digestType>9</secDNS:digestType>
+            <secDNS:digestType>1</secDNS:digestType>
            <secDNS:digest>12AB</secDNS:digest>
          </secDNS:dsData>
          <secDNS:dsData>
@@ -51,13 +51,13 @@
          <secDNS:dsData>
            <secDNS:keyTag>1</secDNS:keyTag>
            <secDNS:alg>2</secDNS:alg>
-            <secDNS:digestType>3</secDNS:digestType>
+            <secDNS:digestType>2</secDNS:digestType>
            <secDNS:digest>ABCD</secDNS:digest>
          </secDNS:dsData>
          <secDNS:dsData>
            <secDNS:keyTag>4</secDNS:keyTag>
            <secDNS:alg>5</secDNS:alg>
-            <secDNS:digestType>6</secDNS:digestType>
+            <secDNS:digestType>1</secDNS:digestType>
            <secDNS:digest>EF01</secDNS:digest>
          </secDNS:dsData>
        </secDNS:add>
@@ -37,7 +37,7 @@
          <secDNS:dsData>
            <secDNS:keyTag>7</secDNS:keyTag>
            <secDNS:alg>8</secDNS:alg>
-            <secDNS:digestType>9</secDNS:digestType>
+            <secDNS:digestType>1</secDNS:digestType>
            <secDNS:digest>12AB</secDNS:digest>
          </secDNS:dsData>
          <secDNS:dsData>
@@ -51,13 +51,13 @@
          <secDNS:dsData>
            <secDNS:keyTag>1</secDNS:keyTag>
            <secDNS:alg>2</secDNS:alg>
-            <secDNS:digestType>3</secDNS:digestType>
+            <secDNS:digestType>2</secDNS:digestType>
            <secDNS:digest>ABCD</secDNS:digest>
          </secDNS:dsData>
          <secDNS:dsData>
            <secDNS:keyTag>4</secDNS:keyTag>
            <secDNS:alg>5</secDNS:alg>
-            <secDNS:digestType>6</secDNS:digestType>
+            <secDNS:digestType>1</secDNS:digestType>
            <secDNS:digest>EF01</secDNS:digest>
          </secDNS:dsData>
        </secDNS:add>
@@ -21,7 +21,7 @@
          <secDNS:dsData>
            <secDNS:keyTag>7</secDNS:keyTag>
            <secDNS:alg>8</secDNS:alg>
-            <secDNS:digestType>9</secDNS:digestType>
+            <secDNS:digestType>1</secDNS:digestType>
            <secDNS:digest>12AB</secDNS:digest>
          </secDNS:dsData>
          <secDNS:dsData>
@@ -16,13 +16,13 @@
          <secDNS:dsData>
            <secDNS:keyTag>1</secDNS:keyTag>
            <secDNS:alg>2</secDNS:alg>
-            <secDNS:digestType>3</secDNS:digestType>
+            <secDNS:digestType>2</secDNS:digestType>
            <secDNS:digest>ABCD</secDNS:digest>
          </secDNS:dsData>
          <secDNS:dsData>
            <secDNS:keyTag>4</secDNS:keyTag>
            <secDNS:alg>5</secDNS:alg>
-            <secDNS:digestType>6</secDNS:digestType>
+            <secDNS:digestType>1</secDNS:digestType>
            <secDNS:digest>EF01</secDNS:digest>
          </secDNS:dsData>
        </secDNS:add>
@@ -261,19 +261,11 @@ td.section {
    </tr> 
    <tr> 
     <td class="property_name">generated on</td> 
-<<<<<<< HEAD
-     <td class="property_value">2021-11-23 20:43:39.672931</td> 
+     <td class="property_value">2022-02-03 18:23:54.985476</td> 
    </tr> 
    <tr>
     <td class="property_name">last flyway file</td>
-     <td id="lastFlywayFile" class="property_value">V104__add_transfer_response_host_id_to_poll_message.sql</td>
-=======
-     <td class="property_value">2021-11-17 21:23:10.198048</td> 
-    </tr> 
-    <tr>
-     <td class="property_name">last flyway file</td>
-     <td id="lastFlywayFile" class="property_value">V103__add_transfer_response_host_id_to_poll_message.sql</td>
->>>>>>> 19aa29939 (Fix missing hostPendingActionNotificationResponses in PollMessage.OneTime)
+     <td id="lastFlywayFile" class="property_value">V105__add_index_on_host_name_in_host_table.sql</td>
    </tr>
   </tbody>
  </table> 
@@ -292,11 +284,7 @@ td.section {
     generated on
    </text> 
    <text text-anchor="start" x="4755.52" y="-10.8" font-family="Helvetica,sans-Serif" font-size="14.00">
-<<<<<<< HEAD
-     2021-11-23 20:43:39.672931
-=======
-     2021-11-17 21:23:10.198048
->>>>>>> 19aa29939 (Fix missing hostPendingActionNotificationResponses in PollMessage.OneTime)
+     2022-02-03 18:23:54.985476
    </text> 
    <polygon fill="none" stroke="#888888" points="4668.02,-4 4668.02,-44 4933.02,-44 4933.02,-4 4668.02,-4" /> <!-- allocationtoken_a08ccbef --> 
    <g id="node1" class="node"> 
@@ -11179,6 +11167,18 @@ td.section {
    <tr> 
     <td colspan="3"></td> 
    </tr> 
+    <tr> 
+     <td colspan="2" class="name">idxkpkh68n6dy5v51047yr6b0e9l</td> 
+     <td class="description right">[non-unique index]</td> 
+    </tr> 
+    <tr> 
+     <td class="spacer"></td> 
+     <td class="minwidth">host_name</td> 
+     <td class="minwidth">ascending</td> 
+    </tr> 
+    <tr> 
+     <td colspan="3"></td> 
+    </tr> 
    <tr> 
     <td colspan="2" class="name">Host_pkey</td> 
     <td class="description right">[unique index]</td> 
@@ -102,3 +102,4 @@ V101__domain_add_dns_refresh_request_time.sql
 V102__add_indexes_to_domain_history_sub_tables.sql
 V103__creation_time_not_null.sql
 V104__add_transfer_response_host_id_to_poll_message.sql
+V105__add_index_on_host_name_in_host_table.sql
@@ -0,0 +1,15 @@
+-- Copyright 2022 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.
+
+CREATE INDEX IDXkpkh68n6dy5v51047yr6b0e9l ON "Host" (host_name);
@@ -804,6 +804,7 @@ create index IDX6w3qbtgce93cal2orjg1tw7b7 on "DomainHistory" (history_modificati
       add constraint UKat9erbh52e4lg3jw6ai9wkjj9 unique (domain_repo_id, host_repo_id);
 create index IDXj1mtx98ndgbtb1bkekahms18w on "GracePeriod" (domain_repo_id);
 create index IDXd01j17vrpjxaerxdmn8bwxs7s on "GracePeriodHistory" (domain_repo_id);
+create index IDXkpkh68n6dy5v51047yr6b0e9l on "Host" (host_name);
 create index IDXfg2nnjlujxo6cb9fha971bq2n on "HostHistory" (creation_time);
 create index IDX1iy7njgb7wjmj9piml4l2g0qi on "HostHistory" (history_registrar_id);
 create index IDXkkwbwcwvrdkkqothkiye4jiff on "HostHistory" (host_name);
@@ -1728,6 +1728,13 @@ CREATE INDEX idxkjt9yaq92876dstimd93hwckh ON public."Domain" USING btree (curren
 CREATE INDEX idxknk8gmj7s47q56cwpa6rmpt5l ON public."HostHistory" USING btree (history_type);


+--
+-- Name: idxkpkh68n6dy5v51047yr6b0e9l; Type: INDEX; Schema: public; Owner: -
+--
+
+CREATE INDEX idxkpkh68n6dy5v51047yr6b0e9l ON public."Host" USING btree (host_name);
+
+
 --
 -- Name: idxlrq7v63pc21uoh3auq6eybyhl; Type: INDEX; Schema: public; Owner: -
 --
@@ -398,6 +398,7 @@ An EPP flow that creates a new domain resource.
    *   The fee description passed in the transform command matches multiple fee
        types.
    *   The fee description passed in the transform command cannot be parsed.
+    *   Domain has an invalid DS record.
    *   Domain name starts with xn-- but is not a valid IDN.
    *   The specified trademark validator is not supported.
    *   Domain labels cannot begin with a dash.
@@ -810,6 +811,7 @@ statuses are updated at once.
 *   2306
    *   Cannot add and remove the same value.
    *   More than one contact for a given role is not allowed.
+    *   Domain has an invalid DS record.
    *   Missing type attribute for contact.
    *   The secDNS:all element must have value 'true' if present.
    *   Too many DS records set on a domain.
@@ -52,8 +52,8 @@ steps:
    do
      for version in $(gcloud app versions list \
        --filter="SERVICE:$service AND SERVING_STATUS:STOPPED" \
-        --sort-by=LAST_DEPLOYED --format="value(VERSION.ID)" \
-        --project=$project_id | head -n -3)
+        --format="value(VERSION.ID,LAST_DEPLOYED)" \
+        --project=$project_id | sort -k 2 | head -n -3)
      do
        gcloud app versions delete $version --service=$service \
          --project=$project_id --quiet;
@@ -43,6 +43,7 @@ import java.util.Arrays;
 import java.util.Optional;
 import java.util.Random;
 import java.util.function.Supplier;
+import org.joda.time.Duration;

 /** Utilities for dealing with Cloud Tasks. */
 public class CloudTasksUtils implements Serializable {
@@ -167,12 +168,45 @@ public class CloudTasksUtils implements Serializable {
    if (!jitterSeconds.isPresent() || jitterSeconds.get() <= 0) {
      return createTask(path, method, service, params);
    }
-    Instant scheduleTime =
-        Instant.ofEpochMilli(
-            clock
-                .nowUtc()
-                .plusMillis(random.nextInt((int) SECONDS.toMillis(jitterSeconds.get())))
-                .getMillis());
+    return createTask(
+        path,
+        method,
+        service,
+        params,
+        clock,
+        Duration.millis(random.nextInt((int) SECONDS.toMillis(jitterSeconds.get()))));
+  }
+
+  /**
+   * Create a {@link Task} to be enqueued with delay of {@code duration}.
+   *
+   * @param path the relative URI (staring with a slash and ending without one).
+   * @param method the HTTP method to be used for the request, only GET and POST are supported.
+   * @param service the App Engine service to route the request to. Note that with App Engine Task
+   *     Queue API if no service is specified, the service which enqueues the task will be used to
+   *     process the task. Cloud Tasks API does not support this feature so the service will always
+   *     needs to be explicitly specified.
+   * @param params a multi-map of URL query parameters. Duplicate keys are saved as is, and it is up
+   *     to the server to process the duplicate keys.
+   * @param clock a source of time.
+   * @param delay the amount of time that a task needs to delayed for.
+   * @return the enqueued task.
+   * @see <a
+   *     href=ttps://cloud.google.com/appengine/docs/standard/java/taskqueue/push/creating-tasks#target>Specifyinig
+   *     the worker service</a>
+   */
+  private static Task createTask(
+      String path,
+      HttpMethod method,
+      String service,
+      Multimap<String, String> params,
+      Clock clock,
+      Duration delay) {
+    if (delay.isEqual(Duration.ZERO)) {
+      return createTask(path, method, service, params);
+    }
+    checkArgument(delay.isLongerThan(Duration.ZERO), "Negative duration is not supported.");
+    Instant scheduleTime = Instant.ofEpochMilli(clock.nowUtc().getMillis() + delay.getMillis());
    return Task.newBuilder(createTask(path, method, service, params))
        .setScheduleTime(
            Timestamp.newBuilder()
@@ -214,6 +248,18 @@ public class CloudTasksUtils implements Serializable {
    return createTask(path, HttpMethod.GET, service, params, clock, jitterSeconds);
  }

+  /** Create a {@link Task} via HTTP.POST that will be delayed for {@code delay}. */
+  public static Task createPostTask(
+      String path, String service, Multimap<String, String> params, Clock clock, Duration delay) {
+    return createTask(path, HttpMethod.POST, service, params, clock, delay);
+  }
+
+  /** Create a {@link Task} via HTTP.GET that will be delayed for {@code delay}. */
+  public static Task createGetTask(
+      String path, String service, Multimap<String, String> params, Clock clock, Duration delay) {
+    return createTask(path, HttpMethod.GET, service, params, clock, delay);
+  }
+
  public abstract static class SerializableCloudTasksClient implements Serializable {
    public abstract Task enqueue(String projectId, String locationId, String queueName, Task task);
  }
@@ -34,6 +34,7 @@ import java.nio.charset.StandardCharsets;
 import java.time.Instant;
 import java.util.Optional;
 import org.joda.time.DateTime;
+import org.joda.time.Duration;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;

@@ -219,6 +220,87 @@ public class CloudTasksUtilsTest {
    assertThat(task.getScheduleTime().getSeconds()).isEqualTo(0);
  }

+  @Test
+  void testSuccess_createGetTasks_withDelay() {
+    Task task =
+        CloudTasksUtils.createGetTask(
+            "/the/path", "myservice", params, clock, Duration.standardMinutes(10));
+    assertThat(task.getAppEngineHttpRequest().getHttpMethod()).isEqualTo(HttpMethod.GET);
+    assertThat(task.getAppEngineHttpRequest().getRelativeUri())
+        .isEqualTo("/the/path?key1=val1&key2=val2&key1=val3");
+    assertThat(task.getAppEngineHttpRequest().getAppEngineRouting().getService())
+        .isEqualTo("myservice");
+    assertThat(Instant.ofEpochSecond(task.getScheduleTime().getSeconds()))
+        .isEqualTo(Instant.ofEpochMilli(clock.nowUtc().plusMinutes(10).getMillis()));
+  }
+
+  @Test
+  void testSuccess_createPostTasks_withDelay() {
+    Task task =
+        CloudTasksUtils.createPostTask(
+            "/the/path", "myservice", params, clock, Duration.standardMinutes(10));
+    assertThat(task.getAppEngineHttpRequest().getHttpMethod()).isEqualTo(HttpMethod.POST);
+    assertThat(task.getAppEngineHttpRequest().getRelativeUri()).isEqualTo("/the/path");
+    assertThat(task.getAppEngineHttpRequest().getAppEngineRouting().getService())
+        .isEqualTo("myservice");
+    assertThat(task.getAppEngineHttpRequest().getHeadersMap().get("Content-Type"))
+        .isEqualTo("application/x-www-form-urlencoded");
+    assertThat(task.getAppEngineHttpRequest().getBody().toString(StandardCharsets.UTF_8))
+        .isEqualTo("key1=val1&key2=val2&key1=val3");
+    assertThat(task.getScheduleTime().getSeconds()).isNotEqualTo(0);
+    assertThat(Instant.ofEpochSecond(task.getScheduleTime().getSeconds()))
+        .isEqualTo(Instant.ofEpochMilli(clock.nowUtc().plusMinutes(10).getMillis()));
+  }
+
+  @Test
+  void testFailure_createGetTasks_withNegativeDelay() {
+    IllegalArgumentException thrown =
+        assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                CloudTasksUtils.createGetTask(
+                    "/the/path", "myservice", params, clock, Duration.standardMinutes(-10)));
+    assertThat(thrown).hasMessageThat().isEqualTo("Negative duration is not supported.");
+  }
+
+  @Test
+  void testFailure_createPostTasks_withNegativeDelay() {
+    IllegalArgumentException thrown =
+        assertThrows(
+            IllegalArgumentException.class,
+            () ->
+                CloudTasksUtils.createGetTask(
+                    "/the/path", "myservice", params, clock, Duration.standardMinutes(-10)));
+    assertThat(thrown).hasMessageThat().isEqualTo("Negative duration is not supported.");
+  }
+
+  @Test
+  void testSuccess_createPostTasks_withZeroDelay() {
+    Task task =
+        CloudTasksUtils.createPostTask("/the/path", "myservice", params, clock, Duration.ZERO);
+    assertThat(task.getAppEngineHttpRequest().getHttpMethod()).isEqualTo(HttpMethod.POST);
+    assertThat(task.getAppEngineHttpRequest().getRelativeUri()).isEqualTo("/the/path");
+    assertThat(task.getAppEngineHttpRequest().getAppEngineRouting().getService())
+        .isEqualTo("myservice");
+    assertThat(task.getAppEngineHttpRequest().getHeadersMap().get("Content-Type"))
+        .isEqualTo("application/x-www-form-urlencoded");
+    assertThat(task.getAppEngineHttpRequest().getBody().toString(StandardCharsets.UTF_8))
+        .isEqualTo("key1=val1&key2=val2&key1=val3");
+    assertThat(task.getScheduleTime().getSeconds()).isEqualTo(0);
+  }
+
+  @Test
+  void testSuccess_createGetTasks_withZeroDelay() {
+    Task task =
+        CloudTasksUtils.createGetTask("/the/path", "myservice", params, clock, Duration.ZERO);
+    assertThat(task.getAppEngineHttpRequest().getHttpMethod()).isEqualTo(HttpMethod.GET);
+    assertThat(task.getAppEngineHttpRequest().getRelativeUri())
+        .isEqualTo("/the/path?key1=val1&key2=val2&key1=val3");
+    assertThat(task.getAppEngineHttpRequest().getAppEngineRouting().getService())
+        .isEqualTo("myservice");
+    assertThat(task.getScheduleTime().getSeconds()).isEqualTo(0);
+  }
+
  @Test
  void testFailure_illegalPath() {
    assertThrows(
Author	SHA1	Message	Date
Weimin Yu	b412bdef9f	Fix flaky RdeStagingActionDatastoreTest (#1514 ) * Fix flaky RdeStagingActionDatastoreTest Fixed the most common cause that makes one method flaky (Clock and timestamp problem). Added a TODO to rethink test case. Also added notes on tasks potentially enqueued multiple times.	2022-02-04 10:40:52 -05:00
Rachel Guan	62e5de8a3a	Add support for delay of duration when scheduling a task (#1493 ) * Add support for delay by duration when scheduling task * Fix comments * Add test for negative duration * Change delay parameter type to duration	2022-02-03 22:25:39 -05:00
Lai Jiang	fa9b784c5c	Correctly delete all stopped versions except for the most recent 3 (#1511 ) The gcloud command does some weird stuff with sorting when custom format is used. Here we instead rely on linux sort and head command to sort the versions list. <!-- Reviewable:start --> --- This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/google/nomulus/1511) <!-- Reviewable:end -->	2022-02-03 16:04:58 -05:00
Weimin Yu	e2bd72a74e	Add an index on Host.host_name column (#1510 ) * Add an index on Host.host_name column This field is queried during host creation and needs an index to speed up the query. Since Hibernate does not explicitly refer to indexes, we can change the code and schema in one PR.	2022-02-03 15:57:15 -05:00
gbrodman	28d41488b1	Use the built-in replicaJpaTm() in RDAP (#1506 ) * Use the built-in replicaJpaTm() in RDAP This includes a test for the replica-simulating transaction manager and removal of any replica-specific code in RDAP tests, because it's unnecessary due to the existing tests.	2022-02-03 11:14:26 -05:00
Weimin Yu	1107b9f2e3	Count duplicates when comparing Databases (#1509 ) * Count duplicates when comparing Databases Cursors may have duplicates in Datastore if imported across projects. Count them instead of throwing.	2022-02-03 10:59:03 -05:00
Lai Jiang	9624b483d4	Copy the latest revision of BRDA during upload (#1508 ) The revision was hardcoded to 0, which caused problem when we need to re-run BRDA. <!-- Reviewable:start --> --- This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/google/nomulus/1508) <!-- Reviewable:end -->	2022-02-02 21:54:42 -05:00
Rachel Guan	365937f22d	Change from TaskQueueUtils to CloudTasksUtils in RdeStaging (#1411 ) * Change from TaskQueueUtils to CloudTaskUtils in RdeStaging	2022-02-01 20:41:56 -05:00
sarahcaseybot	d5db6c16bc	Add DS validation to match Cloud DNS (#1487 ) * Add DS validation to match Cloud DNS * Add checks to flows * Add some flow tests * Add tests for DomainCreateFlow * Add tests for UpdateDomainCommand * Fix docs test * Small fixes * Remove builder from tests	2022-02-01 15:25:00 -05:00
Lai Jiang	c1ad06afd1	Allow the beam parameter in RDE standard mode (#1505 ) Standard mode will determine the watermarks based on the cursors and kick off subsequent uploading steps. In order to run both the Beam and the Mapreduce pipeline in parallel, we need to allow setting the beam parameter when in standard mode. This changes should have been part of https://github.com/google/nomulus/pull/1500. <!-- Reviewable:start --> --- This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/google/nomulus/1505) <!-- Reviewable:end -->	2022-01-31 14:20:23 -05:00