Add support for Nordn upload without using pull queues. (#1925)

This PR adds an alternative method to upload Lordn to Nordn server without
using App Engine pull queue. A new database migration stage is added to control
whether a new task is scheduled with the old or new method. The
NordnUploadAction is configured to process both kind of tasks. Once the tasks
scheduled for the old tasks are all processed, we can start using the
new method exclusively.

See: go/registry-pull-queue-redesign

core/src/main/java/google/registry/env/production/default/WEB-INF/cron.xml

@@ -162,7 +162,7 @@
   </cron>
 
   <cron>
-    <url><![CDATA[/_dr/cron/fanout?queue=nordn&endpoint=/_dr/task/nordnUpload&forEachRealTld&lordn-phase=sunrise]]></url>
+    <url><![CDATA[/_dr/cron/fanout?queue=nordn&endpoint=/_dr/task/nordnUpload&forEachRealTld&lordnPhase=sunrise]]></url>
     <description>
       This job uploads LORDN Sunrise CSV files for each TLD to MarksDB. It should be
       run at most every three hours, or at absolute minimum every 26 hours.
@@ -174,7 +174,7 @@
   </cron>
 
   <cron>
-    <url><![CDATA[/_dr/cron/fanout?queue=nordn&endpoint=/_dr/task/nordnUpload&forEachRealTld&lordn-phase=claims]]></url>
+    <url><![CDATA[/_dr/cron/fanout?queue=nordn&endpoint=/_dr/task/nordnUpload&forEachRealTld&lordnPhase=claims]]></url>
     <description>
       This job uploads LORDN Claims CSV files for each TLD to MarksDB. It should be
       run at most every three hours, or at absolute minimum every 26 hours.
@@ -185,6 +185,32 @@
     <target>backend</target>
   </cron>
 
+  <cron>
+    <url><![CDATA[/_dr/cron/fanout?queue=nordn&endpoint=/_dr/task/nordnUpload&forEachRealTld&lordnPhase=sunrise&pullQueue]]></url>
+    <description>
+      This job uploads LORDN Sunrise CSV files for each TLD to MarksDB using
+      pull queue. It should be run at most every three hours, or at absolute
+      minimum every 26 hours.
+    </description>
+    <!-- This may be set anywhere between "every 3 hours" and "every 25 hours". -->
+    <schedule>every 12 hours synchronized</schedule>
+    <timezone>UTC</timezone>
+    <target>backend</target>
+  </cron>
+
+  <cron>
+    <url><![CDATA[/_dr/cron/fanout?queue=nordn&endpoint=/_dr/task/nordnUpload&forEachRealTld&lordnPhase=claims&pullQueue]]></url>
+    <description>
+      This job uploads LORDN Claims CSV files for each TLD to MarksDB using pull
+      queue. It should be run at most every three hours, or at absolute minimum
+      every 26 hours.
+    </description>
+    <!-- This may be set anywhere between "every 3 hours" and "every 25 hours". -->
+    <schedule>every 12 hours synchronized</schedule>
+    <timezone>UTC</timezone>
+    <target>backend</target>
+  </cron>
+
   <cron>
     <url><![CDATA[/_dr/cron/fanout?queue=retryable-cron-tasks&endpoint=/_dr/task/deleteProberData&runInEmpty]]></url>
     <description>

core/src/main/java/google/registry/flows/domain/DomainCreateFlow.java

@@ -86,6 +86,8 @@ import google.registry.model.billing.BillingEvent.Flag;
 import google.registry.model.billing.BillingEvent.Reason;
 import google.registry.model.billing.BillingEvent.Recurring;
 import google.registry.model.billing.BillingEvent.RenewalPriceBehavior;
+import google.registry.model.common.DatabaseMigrationStateSchedule;
+import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
 import google.registry.model.domain.Domain;
 import google.registry.model.domain.DomainCommand;
 import google.registry.model.domain.DomainCommand.Create;
@@ -123,6 +125,7 @@ import google.registry.model.tmch.ClaimsList;
 import google.registry.model.tmch.ClaimsListDao;
 import google.registry.persistence.VKey;
 import google.registry.tmch.LordnTaskUtils;
+import google.registry.tmch.LordnTaskUtils.LordnPhase;
 import java.util.Map;
 import java.util.Optional;
 import javax.annotation.Nullable;
@@ -377,7 +380,7 @@ public final class DomainCreateFlow implements TransactionalFlow {
         reservationTypes.contains(NAME_COLLISION)
             ? ImmutableSet.of(SERVER_HOLD)
             : ImmutableSet.of();
-    Domain domain =
+    Domain.Builder domainBuilder =
         new Domain.Builder()
             .setCreationRegistrarId(registrarId)
             .setPersistedCurrentSponsorRegistrarId(registrarId)
@@ -397,7 +400,14 @@ public final class DomainCreateFlow implements TransactionalFlow {
             .setContacts(command.getContacts())
             .addGracePeriod(
                 GracePeriod.forBillingEvent(GracePeriodStatus.ADD, repoId, createBillingEvent))
-            .build();
+            .setLordnPhase(
+                !DatabaseMigrationStateSchedule.getValueAtTime(tm().getTransactionTime())
+                        .equals(MigrationState.NORDN_SQL)
+                    ? LordnPhase.NONE
+                    : hasSignedMarks
+                        ? LordnPhase.SUNRISE
+                        : hasClaimsNotice ? LordnPhase.CLAIMS : LordnPhase.NONE);
+    Domain domain = domainBuilder.build();
     if (allocationToken.isPresent()
         && allocationToken.get().getTokenType().equals(TokenType.PACKAGE)) {
       if (years > 1) {
@@ -697,7 +707,9 @@ public final class DomainCreateFlow implements TransactionalFlow {
     if (newDomain.shouldPublishToDns()) {
       dnsQueue.addDomainRefreshTask(newDomain.getDomainName());
     }
-    if (hasClaimsNotice || hasSignedMarks) {
+    if (!DatabaseMigrationStateSchedule.getValueAtTime(tm().getTransactionTime())
+            .equals(MigrationState.NORDN_SQL)
+        && (hasClaimsNotice || hasSignedMarks)) {
       LordnTaskUtils.enqueueDomainTask(newDomain);
     }
   }

core/src/main/java/google/registry/model/common/DatabaseMigrationStateSchedule.java

@@ -44,6 +44,8 @@ public class DatabaseMigrationStateSchedule extends CrossTldSingleton {
 
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
+  private static boolean useUncachedForTest = false;
+
   public enum PrimaryDatabase {
     CLOUD_SQL,
     DATASTORE
@@ -85,7 +87,10 @@ public class DatabaseMigrationStateSchedule extends CrossTldSingleton {
     SQL_ONLY(PrimaryDatabase.CLOUD_SQL, false, ReplayDirection.NO_REPLAY),
 
     /** Toggles SQL Sequence based allocateId */
-    SEQUENCE_BASED_ALLOCATE_ID(PrimaryDatabase.CLOUD_SQL, false, ReplayDirection.NO_REPLAY);
+    SEQUENCE_BASED_ALLOCATE_ID(PrimaryDatabase.CLOUD_SQL, false, ReplayDirection.NO_REPLAY),
+
+    /** Use SQL-based Nordn upload flow instead of the pull queue-based one. */
+    NORDN_SQL(PrimaryDatabase.CLOUD_SQL, false, ReplayDirection.NO_REPLAY);
 
     private final PrimaryDatabase primaryDatabase;
     private final boolean isReadOnly;
@@ -164,7 +169,9 @@ public class DatabaseMigrationStateSchedule extends CrossTldSingleton {
                 MigrationState.SQL_ONLY,
                 MigrationState.SQL_PRIMARY_READ_ONLY,
                 MigrationState.SQL_PRIMARY)
-            .putAll(MigrationState.SQL_ONLY, MigrationState.SEQUENCE_BASED_ALLOCATE_ID);
+            .putAll(MigrationState.SQL_ONLY, MigrationState.SEQUENCE_BASED_ALLOCATE_ID)
+            .putAll(MigrationState.SEQUENCE_BASED_ALLOCATE_ID, MigrationState.NORDN_SQL)
+            .putAll(MigrationState.NORDN_SQL, MigrationState.SEQUENCE_BASED_ALLOCATE_ID);
 
     // In addition, we can always transition from a state to itself (useful when updating the map).
     Arrays.stream(MigrationState.values()).forEach(state -> builder.put(state, state));
@@ -181,8 +188,8 @@ public class DatabaseMigrationStateSchedule extends CrossTldSingleton {
   public TimedTransitionProperty<MigrationState> migrationTransitions =
       TimedTransitionProperty.withInitialValue(MigrationState.DATASTORE_ONLY);
 
-  // Required for Objectify initialization
+  // Required for Hibernate initialization
-  private DatabaseMigrationStateSchedule() {}
+  protected DatabaseMigrationStateSchedule() {}
 
   @VisibleForTesting
   public DatabaseMigrationStateSchedule(
@@ -205,6 +212,11 @@ public class DatabaseMigrationStateSchedule extends CrossTldSingleton {
     CACHE.invalidateAll();
   }
 
+  @VisibleForTesting
+  public static void useUncachedForTest() {
+    useUncachedForTest = true;
+  }
+
   /** Loads the currently-set migration schedule from the cache, or the default if none exists. */
   public static TimedTransitionProperty<MigrationState> get() {
     return CACHE.get(DatabaseMigrationStateSchedule.class);
@@ -212,7 +224,9 @@ public class DatabaseMigrationStateSchedule extends CrossTldSingleton {
 
   /** Returns the database migration status at the given time. */
   public static MigrationState getValueAtTime(DateTime dateTime) {
-    return get().getValueAtTime(dateTime);
+    return useUncachedForTest
+        ? getUncached().getValueAtTime(dateTime)
+        : get().getValueAtTime(dateTime);
   }
 
   /** Loads the currently-set migration schedule from SQL, or the default if none exists. */

core/src/main/java/google/registry/tmch/LordnTaskUtils.java

@@ -66,7 +66,12 @@ public final class LordnTaskUtils {
   }
 
   /** Returns the corresponding CSV LORDN line for a sunrise domain. */
-  public static String getCsvLineForSunriseDomain(Domain domain, DateTime transactionTime) {
+  public static String getCsvLineForSunriseDomain(Domain domain) {
+    return getCsvLineForSunriseDomain(domain, domain.getCreationTime());
+  }
+
+  // TODO: Merge into the function above after pull queue migration.
+  private static String getCsvLineForSunriseDomain(Domain domain, DateTime transactionTime) {
     return Joiner.on(',')
         .join(
             domain.getRepoId(),
@@ -77,7 +82,12 @@ public final class LordnTaskUtils {
   }
 
   /** Returns the corresponding CSV LORDN line for a claims domain. */
-  public static String getCsvLineForClaimsDomain(Domain domain, DateTime transactionTime) {
+  public static String getCsvLineForClaimsDomain(Domain domain) {
+    return getCsvLineForClaimsDomain(domain, domain.getCreationTime());
+  }
+
+  // TODO: Merge into the function above after pull queue migration.
+  private static String getCsvLineForClaimsDomain(Domain domain, DateTime transactionTime) {
     return Joiner.on(',')
         .join(
             domain.getRepoId(),
@@ -100,16 +110,8 @@ public final class LordnTaskUtils {
   private LordnTaskUtils() {}
 
   public enum LordnPhase {
-    SUNRISE(QUEUE_SUNRISE),
+    SUNRISE,
-
+    CLAIMS,
-    CLAIMS(QUEUE_CLAIMS),
+    NONE
-
-    NONE(null);
-
-    final String queue;
-
-    LordnPhase(String queue) {
-      this.queue = queue;
-    }
   }
 }

core/src/main/java/google/registry/tmch/NordnUploadAction.java

@@ -19,9 +19,13 @@ import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.net.HttpHeaders.LOCATION;
 import static com.google.common.net.MediaType.CSV_UTF_8;
+import static google.registry.persistence.transaction.QueryComposer.Comparator.EQ;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.request.UrlConnectionUtils.getResponseBytes;
 import static google.registry.tmch.LordnTaskUtils.COLUMNS_CLAIMS;
 import static google.registry.tmch.LordnTaskUtils.COLUMNS_SUNRISE;
+import static google.registry.tmch.LordnTaskUtils.getCsvLineForClaimsDomain;
+import static google.registry.tmch.LordnTaskUtils.getCsvLineForSunriseDomain;
 import static java.nio.charset.StandardCharsets.US_ASCII;
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static javax.servlet.http.HttpServletResponse.SC_ACCEPTED;
@@ -33,6 +37,7 @@ import com.google.appengine.api.taskqueue.TaskHandle;
 import com.google.appengine.api.taskqueue.TransientFailureException;
 import com.google.apphosting.api.DeadlineExceededException;
 import com.google.cloud.tasks.v2.Task;
+import com.google.common.base.Ascii;
 import com.google.common.base.Joiner;
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
@@ -42,6 +47,7 @@ import com.google.common.collect.Lists;
 import com.google.common.collect.Ordering;
 import com.google.common.flogger.FluentLogger;
 import google.registry.config.RegistryConfig.Config;
+import google.registry.model.domain.Domain;
 import google.registry.request.Action;
 import google.registry.request.Action.Service;
 import google.registry.request.Parameter;
@@ -49,6 +55,7 @@ import google.registry.request.RequestParameters;
 import google.registry.request.UrlConnectionService;
 import google.registry.request.UrlConnectionUtils;
 import google.registry.request.auth.Auth;
+import google.registry.tmch.LordnTaskUtils.LordnPhase;
 import google.registry.util.Clock;
 import google.registry.util.CloudTasksUtils;
 import google.registry.util.Retrier;
@@ -59,6 +66,7 @@ import java.net.URL;
 import java.security.GeneralSecurityException;
 import java.security.SecureRandom;
 import java.util.List;
+import java.util.Optional;
 import java.util.Random;
 import java.util.concurrent.TimeUnit;
 import javax.inject.Inject;
@@ -81,9 +89,11 @@ import org.joda.time.Duration;
 public final class NordnUploadAction implements Runnable {
 
   static final String PATH = "/_dr/task/nordnUpload";
-  static final String LORDN_PHASE_PARAM = "lordn-phase";
+  static final String LORDN_PHASE_PARAM = "lordnPhase";
+  // TODO: Delete after migrating off of pull queue.
+  static final String PULL_QUEUE_PARAM = "pullQueue";
 
-  private static final int QUEUE_BATCH_SIZE = 1000;
+  private static final int BATCH_SIZE = 1000;
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
   private static final Duration LEASE_PERIOD = Duration.standardHours(1);
 
@@ -100,31 +110,102 @@ public final class NordnUploadAction implements Runnable {
   @Inject LordnRequestInitializer lordnRequestInitializer;
   @Inject UrlConnectionService urlConnectionService;
 
-  @Inject @Config("tmchMarksdbUrl") String tmchMarksdbUrl;
+  @Inject
-  @Inject @Parameter(LORDN_PHASE_PARAM) String phase;
+  @Config("tmchMarksdbUrl")
-  @Inject @Parameter(RequestParameters.PARAM_TLD) String tld;
+  String tmchMarksdbUrl;
+
+  @Inject
+  @Parameter(LORDN_PHASE_PARAM)
+  String phase;
+
+  @Inject
+  @Parameter(RequestParameters.PARAM_TLD)
+  String tld;
+
+  @Inject
+  @Parameter(PULL_QUEUE_PARAM)
+  Optional<Boolean> usePullQueue;
 
   @Inject CloudTasksUtils cloudTasksUtils;
 
-  @Inject NordnUploadAction() {}
+  @Inject
+  NordnUploadAction() {}
 
   /**
    * These LORDN parameter names correspond to the relative paths in LORDN URLs and cannot be
    * changed on our end.
    */
-  private static final String PARAM_LORDN_PHASE_SUNRISE = "sunrise";
+  private static final String PARAM_LORDN_PHASE_SUNRISE =
+      Ascii.toLowerCase(LordnPhase.SUNRISE.toString());
 
-  private static final String PARAM_LORDN_PHASE_CLAIMS = "claims";
+  private static final String PARAM_LORDN_PHASE_CLAIMS =
+      Ascii.toLowerCase(LordnPhase.CLAIMS.toString());
 
   /** How long to wait before attempting to verify an upload by fetching the log. */
   private static final Duration VERIFY_DELAY = Duration.standardMinutes(30);
 
   @Override
   public void run() {
-    try {
+    if (usePullQueue.orElse(false)) {
-      processLordnTasks();
+      try {
-    } catch (IOException | GeneralSecurityException e) {
+        processLordnTasks();
-      throw new RuntimeException(e);
+      } catch (IOException | GeneralSecurityException e) {
+        throw new RuntimeException(e);
+      }
+    } else {
+      checkArgument(
+          phase.equals(PARAM_LORDN_PHASE_SUNRISE) || phase.equals(PARAM_LORDN_PHASE_CLAIMS),
+          "Invalid phase specified to NordnUploadAction: %s.",
+          phase);
+      tm().transact(
+              () -> {
+                // Note here that we load all domains pending Nordn in one batch, which should not
+                // be a problem for the rate of domain registration that we see. If we anticipate
+                // a peak in claims during TLD launch (sunrise is NOT first-come-first-serve, so
+                // there should be no expectation of a peak during it), we can consider temporarily
+                // increasing the frequency of Nordn upload to reduce the size of each batch.
+                //
+                // We did not further divide the domains into smaller batches because the
+                // read-upload-write operation per small batch needs to be inside a single
+                // transaction to prevent race conditions, and running several uploads in rapid
+                // sucession will likely overwhelm the MarksDB upload server, which recommands a
+                // maximum upload frequency of every 3 hours.
+                //
+                // See:
+                // https://datatracker.ietf.org/doc/html/draft-ietf-regext-tmch-func-spec-01#section-5.2.3.3
+                List<Domain> domains =
+                    tm().createQueryComposer(Domain.class)
+                        .where("lordnPhase", EQ, LordnPhase.valueOf(Ascii.toUpperCase(phase)))
+                        .where("tld", EQ, tld)
+                        .orderBy("creationTime")
+                        .list();
+                if (domains.isEmpty()) {
+                  return;
+                }
+                StringBuilder csv = new StringBuilder();
+                ImmutableList.Builder<Domain> newDomains = new ImmutableList.Builder<>();
+
+                domains.forEach(
+                    domain -> {
+                      if (phase.equals(PARAM_LORDN_PHASE_SUNRISE)) {
+                        csv.append(getCsvLineForSunriseDomain(domain)).append('\n');
+                      } else {
+                        csv.append(getCsvLineForClaimsDomain(domain)).append('\n');
+                      }
+                      Domain newDomain = domain.asBuilder().setLordnPhase(LordnPhase.NONE).build();
+                      newDomains.add(newDomain);
+                    });
+                String columns =
+                    phase.equals(PARAM_LORDN_PHASE_SUNRISE) ? COLUMNS_SUNRISE : COLUMNS_CLAIMS;
+                String header =
+                    String.format("1,%s,%d\n%s\n", clock.nowUtc(), domains.size(), columns);
+                try {
+                  uploadCsvToLordn(String.format("/LORDN/%s/%s", tld, phase), header + csv);
+                } catch (IOException | GeneralSecurityException e) {
+                  throw new RuntimeException(e);
+                }
+                tm().updateAll(newDomains.build());
+              });
     }
   }
 
@@ -158,7 +239,7 @@ public final class NordnUploadAction implements Runnable {
                   queue.leaseTasks(
                       LeaseOptions.Builder.withTag(tld)
                           .leasePeriod(LEASE_PERIOD.getMillis(), TimeUnit.MILLISECONDS)
-                          .countLimit(QUEUE_BATCH_SIZE)),
+                          .countLimit(BATCH_SIZE)),
               TransientFailureException.class,
               DeadlineExceededException.class);
       if (tasks.isEmpty()) {
@@ -171,7 +252,7 @@ public final class NordnUploadAction implements Runnable {
   private void processLordnTasks() throws IOException, GeneralSecurityException {
     checkArgument(
         phase.equals(PARAM_LORDN_PHASE_SUNRISE) || phase.equals(PARAM_LORDN_PHASE_CLAIMS),
-        "Invalid phase specified to Nordn servlet: %s.",
+        "Invalid phase specified to NordnUploadAction: %s.",
         phase);
     DateTime now = clock.nowUtc();
     Queue queue =
@@ -184,12 +265,12 @@ public final class NordnUploadAction implements Runnable {
     // Note: This upload/task deletion isn't done atomically (it's not clear how one would do so
     // anyway). As a result, it is possible that the upload might succeed yet the deletion of
     // enqueued tasks might fail. If so, this would result in the same lines being uploaded to NORDN
-    // across mulitple uploads. This is probably OK; all that we really cannot have is a missing
+    // across multiple uploads. This is probably OK; all that we really cannot have is a missing
     // line.
     if (!tasks.isEmpty()) {
       String csvData = convertTasksToCsv(tasks, now, columns);
       uploadCsvToLordn(String.format("/LORDN/%s/%s", tld, phase), csvData);
-      Lists.partition(tasks, QUEUE_BATCH_SIZE)
+      Lists.partition(tasks, BATCH_SIZE)
           .forEach(
               batch ->
                   retrier.callWithRetry(

core/src/main/java/google/registry/tmch/TmchModule.java

@@ -16,6 +16,7 @@ package google.registry.tmch;
 
 import static com.google.common.io.Resources.asByteSource;
 import static com.google.common.io.Resources.getResource;
+import static google.registry.request.RequestParameters.extractOptionalBooleanParameter;
 import static google.registry.request.RequestParameters.extractRequiredParameter;
 
 import dagger.Module;
@@ -25,6 +26,7 @@ import google.registry.request.HttpException.BadRequestException;
 import google.registry.request.Parameter;
 import java.net.MalformedURLException;
 import java.net.URL;
+import java.util.Optional;
 import javax.servlet.http.HttpServletRequest;
 import org.bouncycastle.openpgp.PGPPublicKey;
 
@@ -64,4 +66,10 @@ public final class TmchModule {
   static String provideNordnLogId(HttpServletRequest req) {
     return extractRequiredParameter(req, NordnVerifyAction.NORDN_LOG_ID_PARAM);
   }
+
+  @Provides
+  @Parameter(NordnUploadAction.PULL_QUEUE_PARAM)
+  static Optional<Boolean> provideUsePullQueue(HttpServletRequest req) {
+    return extractOptionalBooleanParameter(req, NordnUploadAction.PULL_QUEUE_PARAM);
+  }
 }

core/src/main/java/google/registry/tools/GenerateLordnCommand.java

@@ -94,10 +94,10 @@ final class GenerateLordnCommand implements Command {
       Domain domain) {
     String status = " ";
     if (domain.getLaunchNotice() == null && domain.getSmdId() != null) {
-      sunriseCsv.add(LordnTaskUtils.getCsvLineForSunriseDomain(domain, domain.getCreationTime()));
+      sunriseCsv.add(LordnTaskUtils.getCsvLineForSunriseDomain(domain));
       status = "S";
     } else if (domain.getLaunchNotice() != null || domain.getSmdId() != null) {
-      claimsCsv.add(LordnTaskUtils.getCsvLineForClaimsDomain(domain, domain.getCreationTime()));
+      claimsCsv.add(LordnTaskUtils.getCsvLineForClaimsDomain(domain));
       status = "C";
     }
     System.out.printf("%s[%s] ", domain.getDomainName(), status);

core/src/test/java/google/registry/flows/domain/DomainCreateFlowTest.java

@@ -155,6 +155,8 @@ import google.registry.model.billing.BillingEvent;
 import google.registry.model.billing.BillingEvent.Flag;
 import google.registry.model.billing.BillingEvent.Reason;
 import google.registry.model.billing.BillingEvent.RenewalPriceBehavior;
+import google.registry.model.common.DatabaseMigrationStateSchedule;
+import google.registry.model.common.DatabaseMigrationStateSchedule.MigrationState;
 import google.registry.model.domain.Domain;
 import google.registry.model.domain.DomainHistory;
 import google.registry.model.domain.GracePeriod;
@@ -182,6 +184,7 @@ import google.registry.persistence.VKey;
 import google.registry.testing.DatabaseHelper;
 import google.registry.testing.TaskQueueExtension;
 import google.registry.testing.TaskQueueHelper.TaskMatcher;
+import google.registry.tmch.LordnTaskUtils.LordnPhase;
 import google.registry.tmch.SmdrlCsvParser;
 import google.registry.tmch.TmchData;
 import google.registry.tmch.TmchTestData;
@@ -192,9 +195,12 @@ import javax.annotation.Nullable;
 import org.joda.money.Money;
 import org.joda.time.DateTime;
 import org.joda.time.Duration;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.junitpioneer.jupiter.cartesian.CartesianTest;
 import org.junitpioneer.jupiter.cartesian.CartesianTest.Values;
 
@@ -224,6 +230,11 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
     clock.setTo(DateTime.parse("1999-04-03T22:00:00.0Z").minus(Duration.millis(1)));
   }
 
+  @BeforeAll
+  static void beforeAll() {
+    DatabaseMigrationStateSchedule.useUncachedForTest();
+  }
+
   @BeforeEach
   void initCreateTest() throws Exception {
     createTld("tld");
@@ -390,7 +401,10 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
         .that(reloadResourceByForeignKey())
         .hasSmdId(null)
         .and()
-        .hasLaunchNotice(null);
+        .hasLaunchNotice(null)
+        .and()
+        .hasLordnPhase(LordnPhase.NONE);
+    assertNoTasksEnqueued(QUEUE_CLAIMS, QUEUE_SUNRISE);
     assertNoTasksEnqueued(QUEUE_CLAIMS, QUEUE_SUNRISE);
   }
 
@@ -400,14 +414,20 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
         .hasSmdId(SMD_ID)
         .and()
         .hasLaunchNotice(null);
-    String expectedPayload =
+    if (DatabaseMigrationStateSchedule.getValueAtTime(clock.nowUtc())
-        String.format(
+        .equals(MigrationState.NORDN_SQL)) {
-            "%s,%s,%s,1,%s",
+      assertAboutDomains().that(reloadResourceByForeignKey()).hasLordnPhase(LordnPhase.SUNRISE);
-            reloadResourceByForeignKey().getRepoId(),
+    } else {
-            domainName,
+      String expectedPayload =
-            SMD_ID,
+          String.format(
-            SMD_VALID_TIME.plusMillis(17));
+              "%s,%s,%s,1,%s",
-    assertTasksEnqueued(QUEUE_SUNRISE, new TaskMatcher().payload(expectedPayload));
+              reloadResourceByForeignKey().getRepoId(),
+              domainName,
+              SMD_ID,
+              SMD_VALID_TIME.plusMillis(17));
+      assertTasksEnqueued(QUEUE_SUNRISE, new TaskMatcher().payload(expectedPayload));
+      assertAboutDomains().that(reloadResourceByForeignKey()).hasLordnPhase(LordnPhase.NONE);
+    }
   }
 
   private void assertClaimsLordn() throws Exception {
@@ -421,13 +441,19 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
                 "tmch",
                 DateTime.parse("2010-08-16T09:00:00.0Z"),
                 DateTime.parse("2009-08-16T09:00:00.0Z")));
-    TaskMatcher task =
+    if (DatabaseMigrationStateSchedule.getValueAtTime(clock.nowUtc())
-        new TaskMatcher()
+        .equals(MigrationState.NORDN_SQL)) {
-            .payload(
+      assertAboutDomains().that(reloadResourceByForeignKey()).hasLordnPhase(LordnPhase.CLAIMS);
-                reloadResourceByForeignKey().getRepoId()
+    } else {
-                    + ",example-one.tld,370d0b7c9223372036854775807,1,"
+      TaskMatcher task =
-                    + "2009-08-16T09:00:00.017Z,2009-08-16T09:00:00.000Z");
+          new TaskMatcher()
-    assertTasksEnqueued(QUEUE_CLAIMS, task);
+              .payload(
+                  reloadResourceByForeignKey().getRepoId()
+                      + ",example-one.tld,370d0b7c9223372036854775807,1,"
+                      + "2009-08-16T09:00:00.017Z,2009-08-16T09:00:00.000Z");
+      assertTasksEnqueued(QUEUE_CLAIMS, task);
+      assertAboutDomains().that(reloadResourceByForeignKey()).hasLordnPhase(LordnPhase.NONE);
+    }
   }
 
   private void doSuccessfulTest(
@@ -939,8 +965,12 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
     assertAboutEppExceptions().that(thrown).marshalsToXml();
   }
 
-  @Test
+  @ParameterizedTest
-  void testSuccess_claimsNotice() throws Exception {
+  @ValueSource(booleans = {true, false})
+  void testSuccess_claimsNotice(boolean usePullQueue) throws Exception {
+    if (!usePullQueue) {
+      useNordnSql();
+    }
     clock.setTo(DateTime.parse("2009-08-16T09:00:00.0Z"));
     setEppInput("domain_create_claim_notice.xml");
     persistContactsAndHosts();
@@ -950,8 +980,12 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
     assertClaimsLordn();
   }
 
-  @Test
+  @ParameterizedTest
-  void testSuccess_claimsNoticeInQuietPeriod() throws Exception {
+  @ValueSource(booleans = {true, false})
+  void testSuccess_claimsNoticeInQuietPeriod(boolean usePullQueue) throws Exception {
+    if (!usePullQueue) {
+      useNordnSql();
+    }
     allocationToken =
         persistResource(
             new AllocationToken.Builder()
@@ -1220,8 +1254,12 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
     assertAllocationTokenWasRedeemed("abcDEF23456");
   }
 
-  @Test
+  @ParameterizedTest
-  void testSuccess_anchorTenant_withClaims() throws Exception {
+  @ValueSource(booleans = {true, false})
+  void testSuccess_anchorTenant_withClaims(boolean usePullQueue) throws Exception {
+    if (!usePullQueue) {
+      useNordnSql();
+    }
     persistResource(
         new AllocationToken.Builder()
             .setDomainName("example-one.tld")
@@ -1289,8 +1327,12 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
     assertSuccessfulCreate("tld", ImmutableSet.of(SUNRISE, ANCHOR_TENANT));
   }
 
-  @Test
+  @ParameterizedTest
-  void testSuccess_anchorTenantInSunrise_withSignedMark() throws Exception {
+  @ValueSource(booleans = {true, false})
+  void testSuccess_anchorTenantInSunrise_withSignedMark(boolean usePullQueue) throws Exception {
+    if (!usePullQueue) {
+      useNordnSql();
+    }
     allocationToken =
         persistResource(
             new AllocationToken.Builder()
@@ -1890,9 +1932,13 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
     assertSuccessfulCreate("tld", ImmutableSet.of(RESERVED));
   }
 
-  @Test
+  @ParameterizedTest
-  void testSuccess_reservedNameCollisionDomain_inSunrise_setsServerHoldAndPollMessage()
+  @ValueSource(booleans = {true, false})
-      throws Exception {
+  void testSuccess_reservedNameCollisionDomain_inSunrise_setsServerHoldAndPollMessage(
+      boolean usePullQueue) throws Exception {
+    if (!usePullQueue) {
+      useNordnSql();
+    }
     persistResource(
         Registry.get("tld")
             .asBuilder()
@@ -2435,8 +2481,13 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
         "tld", "domain_create_response.xml", SUPERUSER, ImmutableMap.of("DOMAIN", "example.tld"));
   }
 
-  @Test
+  @ParameterizedTest
-  void testSuccess_startDateSunriseRegistration_withEncodedSignedMark() throws Exception {
+  @ValueSource(booleans = {true, false})
+  void testSuccess_startDateSunriseRegistration_withEncodedSignedMark(boolean usePullQueue)
+      throws Exception {
+    if (!usePullQueue) {
+      useNordnSql();
+    }
     createTld("tld", START_DATE_SUNRISE);
     clock.setTo(SMD_VALID_TIME);
     setEppInput(
@@ -2458,8 +2509,13 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
   }
 
   /** Test that missing type= argument on launch create works in start-date sunrise. */
-  @Test
+  @ParameterizedTest
-  void testSuccess_startDateSunriseRegistration_withEncodedSignedMark_noType() throws Exception {
+  @ValueSource(booleans = {true, false})
+  void testSuccess_startDateSunriseRegistration_withEncodedSignedMark_noType(boolean usePullQueue)
+      throws Exception {
+    if (!usePullQueue) {
+      useNordnSql();
+    }
     createTld("tld", START_DATE_SUNRISE);
     clock.setTo(SMD_VALID_TIME);
     setEppInput(
@@ -3512,4 +3568,97 @@ class DomainCreateFlowTest extends ResourceFlowTestCase<DomainCreateFlow, Domain
         .isEqualTo(
             "The package token abc123 cannot be used to register names for longer than 1 year.");
   }
+
+  private static void useNordnSql() {
+    tm().transact(
+            () ->
+                DatabaseMigrationStateSchedule.set(
+                    new ImmutableSortedMap.Builder<DateTime, MigrationState>(Ordering.natural())
+                        .put(START_OF_TIME, MigrationState.DATASTORE_ONLY)
+                        .put(START_OF_TIME.plusMillis(1), MigrationState.DATASTORE_PRIMARY)
+                        .build()));
+    tm().transact(
+            () ->
+                DatabaseMigrationStateSchedule.set(
+                    new ImmutableSortedMap.Builder<DateTime, MigrationState>(Ordering.natural())
+                        .put(START_OF_TIME, MigrationState.DATASTORE_ONLY)
+                        .put(START_OF_TIME.plusMillis(1), MigrationState.DATASTORE_PRIMARY)
+                        .put(START_OF_TIME.plusMillis(2), MigrationState.DATASTORE_PRIMARY_NO_ASYNC)
+                        .build()));
+    tm().transact(
+            () ->
+                DatabaseMigrationStateSchedule.set(
+                    new ImmutableSortedMap.Builder<DateTime, MigrationState>(Ordering.natural())
+                        .put(START_OF_TIME, MigrationState.DATASTORE_ONLY)
+                        .put(START_OF_TIME.plusMillis(1), MigrationState.DATASTORE_PRIMARY)
+                        .put(START_OF_TIME.plusMillis(2), MigrationState.DATASTORE_PRIMARY_NO_ASYNC)
+                        .put(
+                            START_OF_TIME.plusMillis(3), MigrationState.DATASTORE_PRIMARY_READ_ONLY)
+                        .build()));
+    tm().transact(
+            () ->
+                DatabaseMigrationStateSchedule.set(
+                    new ImmutableSortedMap.Builder<DateTime, MigrationState>(Ordering.natural())
+                        .put(START_OF_TIME, MigrationState.DATASTORE_ONLY)
+                        .put(START_OF_TIME.plusMillis(1), MigrationState.DATASTORE_PRIMARY)
+                        .put(START_OF_TIME.plusMillis(2), MigrationState.DATASTORE_PRIMARY_NO_ASYNC)
+                        .put(
+                            START_OF_TIME.plusMillis(3), MigrationState.DATASTORE_PRIMARY_READ_ONLY)
+                        .put(START_OF_TIME.plusMillis(4), MigrationState.SQL_PRIMARY_READ_ONLY)
+                        .build()));
+    tm().transact(
+            () ->
+                DatabaseMigrationStateSchedule.set(
+                    new ImmutableSortedMap.Builder<DateTime, MigrationState>(Ordering.natural())
+                        .put(START_OF_TIME, MigrationState.DATASTORE_ONLY)
+                        .put(START_OF_TIME.plusMillis(1), MigrationState.DATASTORE_PRIMARY)
+                        .put(START_OF_TIME.plusMillis(2), MigrationState.DATASTORE_PRIMARY_NO_ASYNC)
+                        .put(
+                            START_OF_TIME.plusMillis(3), MigrationState.DATASTORE_PRIMARY_READ_ONLY)
+                        .put(START_OF_TIME.plusMillis(4), MigrationState.SQL_PRIMARY_READ_ONLY)
+                        .put(START_OF_TIME.plusMillis(5), MigrationState.SQL_PRIMARY)
+                        .build()));
+    tm().transact(
+            () ->
+                DatabaseMigrationStateSchedule.set(
+                    new ImmutableSortedMap.Builder<DateTime, MigrationState>(Ordering.natural())
+                        .put(START_OF_TIME, MigrationState.DATASTORE_ONLY)
+                        .put(START_OF_TIME.plusMillis(1), MigrationState.DATASTORE_PRIMARY)
+                        .put(START_OF_TIME.plusMillis(2), MigrationState.DATASTORE_PRIMARY_NO_ASYNC)
+                        .put(
+                            START_OF_TIME.plusMillis(3), MigrationState.DATASTORE_PRIMARY_READ_ONLY)
+                        .put(START_OF_TIME.plusMillis(4), MigrationState.SQL_PRIMARY_READ_ONLY)
+                        .put(START_OF_TIME.plusMillis(5), MigrationState.SQL_PRIMARY)
+                        .put(START_OF_TIME.plusMillis(6), MigrationState.SQL_ONLY)
+                        .build()));
+    tm().transact(
+            () ->
+                DatabaseMigrationStateSchedule.set(
+                    new ImmutableSortedMap.Builder<DateTime, MigrationState>(Ordering.natural())
+                        .put(START_OF_TIME, MigrationState.DATASTORE_ONLY)
+                        .put(START_OF_TIME.plusMillis(1), MigrationState.DATASTORE_PRIMARY)
+                        .put(START_OF_TIME.plusMillis(2), MigrationState.DATASTORE_PRIMARY_NO_ASYNC)
+                        .put(
+                            START_OF_TIME.plusMillis(3), MigrationState.DATASTORE_PRIMARY_READ_ONLY)
+                        .put(START_OF_TIME.plusMillis(4), MigrationState.SQL_PRIMARY_READ_ONLY)
+                        .put(START_OF_TIME.plusMillis(5), MigrationState.SQL_PRIMARY)
+                        .put(START_OF_TIME.plusMillis(6), MigrationState.SQL_ONLY)
+                        .put(START_OF_TIME.plusMillis(7), MigrationState.SEQUENCE_BASED_ALLOCATE_ID)
+                        .build()));
+    tm().transact(
+            () ->
+                DatabaseMigrationStateSchedule.set(
+                    new ImmutableSortedMap.Builder<DateTime, MigrationState>(Ordering.natural())
+                        .put(START_OF_TIME, MigrationState.DATASTORE_ONLY)
+                        .put(START_OF_TIME.plusMillis(1), MigrationState.DATASTORE_PRIMARY)
+                        .put(START_OF_TIME.plusMillis(2), MigrationState.DATASTORE_PRIMARY_NO_ASYNC)
+                        .put(
+                            START_OF_TIME.plusMillis(3), MigrationState.DATASTORE_PRIMARY_READ_ONLY)
+                        .put(START_OF_TIME.plusMillis(4), MigrationState.SQL_PRIMARY_READ_ONLY)
+                        .put(START_OF_TIME.plusMillis(5), MigrationState.SQL_PRIMARY)
+                        .put(START_OF_TIME.plusMillis(6), MigrationState.SQL_ONLY)
+                        .put(START_OF_TIME.plusMillis(7), MigrationState.SEQUENCE_BASED_ALLOCATE_ID)
+                        .put(START_OF_TIME.plusMillis(8), MigrationState.NORDN_SQL)
+                        .build()));
+  }
 }

core/src/test/java/google/registry/testing/DatabaseHelper.java

@@ -312,6 +312,7 @@ public final class DatabaseHelper {
     return persistResource(domain.asBuilder().setDeletionTime(deletionTime).build());
   }
 
+  // TODO: delete after pull queue migration.
   /** Persists a domain and enqueues a LORDN task of the appropriate type for it. */
   public static Domain persistDomainAndEnqueueLordn(final Domain domain) {
     final Domain persistedDomain = persistResource(domain);

core/src/test/java/google/registry/testing/DomainSubject.java

@@ -27,6 +27,7 @@ import google.registry.model.domain.launch.LaunchNotice;
 import google.registry.model.domain.secdns.DomainDsData;
 import google.registry.model.eppcommon.AuthInfo;
 import google.registry.testing.TruthChainer.And;
+import google.registry.tmch.LordnTaskUtils.LordnPhase;
 import java.util.Set;
 import org.joda.time.DateTime;
 
@@ -41,7 +42,7 @@ public final class DomainSubject extends AbstractEppResourceSubject<Domain, Doma
   }
 
   public And<DomainSubject> hasDomainName(String domainName) {
-    return hasValue(domainName, actual.getDomainName(), "has domainName");
+    return hasValue(domainName, actual.getDomainName(), "domainName");
   }
 
   public And<DomainSubject> hasExactlyDsData(DomainDsData... dsData) {
@@ -49,15 +50,19 @@ public final class DomainSubject extends AbstractEppResourceSubject<Domain, Doma
   }
 
   public And<DomainSubject> hasExactlyDsData(Set<DomainDsData> dsData) {
-    return hasValue(dsData, actual.getDsData(), "has dsData");
+    return hasValue(dsData, actual.getDsData(), "dsData");
   }
 
   public And<DomainSubject> hasNumDsData(int num) {
-    return hasValue(num, actual.getDsData().size(), "has num dsData");
+    return hasValue(num, actual.getDsData().size(), "dsData.size()");
   }
 
   public And<DomainSubject> hasLaunchNotice(LaunchNotice launchNotice) {
-    return hasValue(launchNotice, actual.getLaunchNotice(), "has launchNotice");
+    return hasValue(launchNotice, actual.getLaunchNotice(), "launchNotice");
+  }
+
+  public And<DomainSubject> hasLordnPhase(LordnPhase lordnPhase) {
+    return hasValue(lordnPhase, actual.getLordnPhase(), "lordnPhase");
   }
 
   public And<DomainSubject> hasAuthInfoPwd(String pw) {
@@ -67,7 +72,7 @@ public final class DomainSubject extends AbstractEppResourceSubject<Domain, Doma
 
   public And<DomainSubject> hasCurrentSponsorRegistrarId(String registrarId) {
     return hasValue(
-        registrarId, actual.getCurrentSponsorRegistrarId(), "has currentSponsorRegistrarId");
+        registrarId, actual.getCurrentSponsorRegistrarId(), "currentSponsorRegistrarId");
   }
 
   public And<DomainSubject> hasRegistrationExpirationTime(DateTime expiration) {

core/src/test/java/google/registry/tmch/NordnUploadActionTest.java

@@ -14,13 +14,17 @@
 
 package google.registry.tmch;
 
+import static com.google.appengine.api.taskqueue.QueueFactory.getQueue;
 import static com.google.common.net.HttpHeaders.AUTHORIZATION;
 import static com.google.common.net.HttpHeaders.CONTENT_TYPE;
 import static com.google.common.net.HttpHeaders.LOCATION;
 import static com.google.common.net.MediaType.FORM_DATA;
 import static com.google.common.truth.Truth.assertThat;
+import static google.registry.model.ForeignKeyUtils.load;
 import static google.registry.testing.DatabaseHelper.createTld;
+import static google.registry.testing.DatabaseHelper.loadByKey;
 import static google.registry.testing.DatabaseHelper.loadRegistrar;
+import static google.registry.testing.DatabaseHelper.newDomain;
 import static google.registry.testing.DatabaseHelper.persistDomainAndEnqueueLordn;
 import static google.registry.testing.DatabaseHelper.persistResource;
 import static java.nio.charset.StandardCharsets.UTF_8;
@@ -34,6 +38,7 @@ import static org.mockito.ArgumentMatchers.startsWith;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 
 import com.google.appengine.api.taskqueue.LeaseOptions;
@@ -48,15 +53,17 @@ import com.google.common.collect.ImmutableList;
 import google.registry.model.domain.Domain;
 import google.registry.model.domain.launch.LaunchNotice;
 import google.registry.model.tld.Registry;
+import google.registry.persistence.VKey;
 import google.registry.persistence.transaction.JpaTestExtensions;
 import google.registry.persistence.transaction.JpaTestExtensions.JpaIntegrationTestExtension;
+import google.registry.request.RequestParameters;
 import google.registry.testing.CloudTasksHelper;
 import google.registry.testing.CloudTasksHelper.TaskMatcher;
-import google.registry.testing.DatabaseHelper;
 import google.registry.testing.FakeClock;
 import google.registry.testing.FakeSleeper;
 import google.registry.testing.FakeUrlConnectionService;
 import google.registry.testing.TaskQueueExtension;
+import google.registry.tmch.LordnTaskUtils.LordnPhase;
 import google.registry.util.CloudTasksUtils;
 import google.registry.util.Retrier;
 import google.registry.util.UrlConnectionException;
@@ -67,25 +74,32 @@ import java.net.URL;
 import java.security.SecureRandom;
 import java.util.List;
 import java.util.Optional;
+import java.util.concurrent.TimeUnit;
 import org.joda.time.DateTime;
+import org.joda.time.Duration;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 /** Unit tests for {@link NordnUploadAction}. */
 class NordnUploadActionTest {
 
   private static final String CLAIMS_CSV =
-      "1,2010-05-01T10:11:12.000Z,1\n"
+      "1,2010-05-04T10:11:12.000Z,2\n"
           + "roid,domain-name,notice-id,registrar-id,registration-datetime,ack-datetime,"
           + "application-datetime\n"
-          + "2-TLD,claims-landrush1.tld,landrush1tcn,99999,2010-05-01T10:11:12.000Z,"
+          + "6-TLD,claims-landrush2.tld,landrush2tcn,88888,2010-05-03T10:11:12.000Z,"
-          + "1969-12-31T23:00:00.000Z\n";
+          + "2010-05-03T08:11:12.000Z\n"
+          + "8-TLD,claims-landrush1.tld,landrush1tcn,99999,2010-05-04T10:11:12.000Z,"
+          + "2010-05-04T09:11:12.000Z\n";
 
   private static final String SUNRISE_CSV =
-      "1,2010-05-01T10:11:12.000Z,1\n"
+      "1,2010-05-04T10:11:12.000Z,2\n"
           + "roid,domain-name,SMD-id,registrar-id,registration-datetime,application-datetime\n"
-          + "2-TLD,sunrise1.tld,my-smdid,99999,2010-05-01T10:11:12.000Z\n";
+          + "2-TLD,sunrise2.tld,new-smdid,88888,2010-05-01T10:11:12.000Z\n"
+          + "4-TLD,sunrise1.tld,my-smdid,99999,2010-05-02T10:11:12.000Z\n";
 
   private static final String LOCATION_URL = "http://trololol";
 
@@ -116,8 +130,12 @@ class NordnUploadActionTest {
     when(httpUrlConnection.getHeaderField(LOCATION)).thenReturn("http://trololol");
     when(httpUrlConnection.getOutputStream()).thenReturn(connectionOutputStream);
     persistResource(loadRegistrar("TheRegistrar").asBuilder().setIanaIdentifier(99999L).build());
+    persistResource(loadRegistrar("NewRegistrar").asBuilder().setIanaIdentifier(88888L).build());
     createTld("tld");
     persistResource(Registry.get("tld").asBuilder().setLordnUsername("lolcat").build());
+    persistSunriseModeDomain();
+    clock.advanceBy(Duration.standardDays(1));
+    persistClaimsModeDomain();
     action.clock = clock;
     action.cloudTasksUtils = cloudTasksUtils;
     action.urlConnectionService = urlConnectionService;
@@ -127,6 +145,7 @@ class NordnUploadActionTest {
     action.tmchMarksdbUrl = "http://127.0.0.1";
     action.random = new SecureRandom();
     action.retrier = new Retrier(new FakeSleeper(clock), 3);
+    action.usePullQueue = Optional.empty();
   }
 
   @Test
@@ -137,7 +156,7 @@ class NordnUploadActionTest {
             makeTaskHandle("task1", "example", "csvLine1", "lordn-sunrise"),
             makeTaskHandle("task3", "example", "ending", "lordn-sunrise"));
     assertThat(NordnUploadAction.convertTasksToCsv(tasks, clock.nowUtc(), "col1,col2"))
-        .isEqualTo("1,2010-05-01T10:11:12.000Z,3\ncol1,col2\ncsvLine1\ncsvLine2\nending\n");
+        .isEqualTo("1,2010-05-04T10:11:12.000Z,3\ncol1,col2\ncsvLine1\ncsvLine2\nending\n");
   }
 
   @Test
@@ -149,13 +168,13 @@ class NordnUploadActionTest {
             makeTaskHandle("task3", "example", "ending", "lordn-sunrise"),
             makeTaskHandle("task1", "example", "csvLine1", "lordn-sunrise"));
     assertThat(NordnUploadAction.convertTasksToCsv(tasks, clock.nowUtc(), "col1,col2"))
-        .isEqualTo("1,2010-05-01T10:11:12.000Z,3\ncol1,col2\ncsvLine1\ncsvLine2\nending\n");
+        .isEqualTo("1,2010-05-04T10:11:12.000Z,3\ncol1,col2\ncsvLine1\ncsvLine2\nending\n");
   }
 
   @Test
   void test_convertTasksToCsv_doesntFailOnEmptyTasks() {
     assertThat(NordnUploadAction.convertTasksToCsv(ImmutableList.of(), clock.nowUtc(), "col1,col2"))
-        .isEqualTo("1,2010-05-01T10:11:12.000Z,0\ncol1,col2\n");
+        .isEqualTo("1,2010-05-04T10:11:12.000Z,0\ncol1,col2\n");
   }
 
   @Test
@@ -187,122 +206,105 @@ class NordnUploadActionTest {
   }
 
   @Test
-  void testRun_claimsMode_appendsTldAndClaimsToRequestUrl() throws Exception {
+  void testSuccess_noPassword_doesntSendAuthorizationHeader() {
-    persistClaimsModeDomain();
-    action.run();
-    assertThat(httpUrlConnection.getURL()).isEqualTo(new URL("http://127.0.0.1/LORDN/tld/claims"));
-  }
-
-  @Test
-  void testRun_sunriseMode_appendsTldAndClaimsToRequestUrl() throws Exception {
-    persistSunriseModeDomain();
-    action.run();
-    assertThat(httpUrlConnection.getURL()).isEqualTo(new URL("http://127.0.0.1/LORDN/tld/sunrise"));
-  }
-
-  @Test
-  void testRun_usesMultipartContentType() throws Exception {
-    persistClaimsModeDomain();
-    action.run();
-    verify(httpUrlConnection)
-        .setRequestProperty(eq(CONTENT_TYPE), startsWith("multipart/form-data; boundary="));
-    verify(httpUrlConnection).setRequestMethod("POST");
-  }
-
-  @Test
-  void testRun_hasPassword_setsAuthorizationHeader() {
-    persistClaimsModeDomain();
-    action.run();
-    verify(httpUrlConnection)
-        .setRequestProperty(
-            AUTHORIZATION, "Basic bG9sY2F0OmF0dGFjaw=="); // echo -n lolcat:attack | base64
-  }
-
-  @Test
-  void testRun_noPassword_doesntSendAuthorizationHeader() {
     action.lordnRequestInitializer = new LordnRequestInitializer(Optional.empty());
-    persistClaimsModeDomain();
     action.run();
     verify(httpUrlConnection, times(0)).setRequestProperty(eq(AUTHORIZATION), anyString());
   }
 
   @Test
-  void testRun_claimsMode_payloadMatchesClaimsCsv() {
+  void testSuccess_nothingScheduled() {
-    persistClaimsModeDomain();
+    persistResource(
+        loadByKey(load(Domain.class, "claims-landrush1.tld", clock.nowUtc()))
+            .asBuilder()
+            .setLordnPhase(LordnPhase.NONE)
+            .build());
+    persistResource(
+        loadByKey(load(Domain.class, "claims-landrush2.tld", clock.nowUtc()))
+            .asBuilder()
+            .setLordnPhase(LordnPhase.NONE)
+            .build());
     action.run();
-    assertThat(connectionOutputStream.toString(UTF_8)).contains(CLAIMS_CSV);
+    verifyNoInteractions(httpUrlConnection);
+    cloudTasksHelper.assertNoTasksEnqueued(NordnVerifyAction.QUEUE);
+  }
+
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
+  void testSuccess_claimsMode(boolean usePullQueue) throws Exception {
+    testRun("claims", "claims-landrush1.tld", "claims-landrush2.tld", CLAIMS_CSV, usePullQueue);
+  }
+
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
+  void testSuccess_sunriseMode(boolean usePullQueue) throws Exception {
+    testRun("sunrise", "sunrise1.tld", "sunrise2.tld", SUNRISE_CSV, usePullQueue);
   }
 
   @Test
-  void testRun_claimsMode_verifyTaskGetsEnqueuedWithClaimsCsv() {
+  void testSuccess_noResponseContent_stillWorksNormally() throws Exception {
-    persistClaimsModeDomain();
-    action.run();
-    cloudTasksHelper.assertTasksEnqueued(
-        NordnVerifyAction.QUEUE,
-        new TaskMatcher()
-            .url(NordnVerifyAction.PATH)
-            .param(NordnVerifyAction.NORDN_URL_PARAM, LOCATION_URL)
-            .header(CONTENT_TYPE, FORM_DATA.toString()));
-  }
-
-  @Test
-  void testRun_sunriseMode_payloadMatchesSunriseCsv() {
-    persistSunriseModeDomain();
-    action.run();
-    assertThat(connectionOutputStream.toString(UTF_8)).contains(SUNRISE_CSV);
-  }
-
-  @Test
-  void test_noResponseContent_stillWorksNormally() throws Exception {
     // Returning null only affects logging.
     when(httpUrlConnection.getInputStream()).thenReturn(new ByteArrayInputStream(new byte[] {}));
-    persistSunriseModeDomain();
+    testRun("claims", "claims-landrush1.tld", "claims-landrush2.tld", CLAIMS_CSV, false);
-    action.run();
-    assertThat(connectionOutputStream.toString(UTF_8)).contains(SUNRISE_CSV);
-  }
-
-  @Test
-  void testRun_sunriseMode_verifyTaskGetsEnqueuedWithSunriseCsv() {
-    persistSunriseModeDomain();
-    action.run();
-    cloudTasksHelper.assertTasksEnqueued(
-        NordnVerifyAction.QUEUE,
-        new TaskMatcher()
-            .url(NordnVerifyAction.PATH)
-            .param(NordnVerifyAction.NORDN_URL_PARAM, LOCATION_URL)
-            .header(CONTENT_TYPE, FORM_DATA.toString()));
   }
 
   @Test
   void testFailure_nullRegistryUser() {
-    persistClaimsModeDomain();
     persistResource(Registry.get("tld").asBuilder().setLordnUsername(null).build());
     VerifyException thrown = assertThrows(VerifyException.class, action::run);
     assertThat(thrown).hasMessageThat().contains("lordnUsername is not set for tld.");
   }
 
   @Test
-  void testFetchFailure() throws Exception {
+  void testFailure_errorResponseCode() throws Exception {
-    persistClaimsModeDomain();
     when(httpUrlConnection.getResponseCode()).thenReturn(SC_INTERNAL_SERVER_ERROR);
     assertThrows(UrlConnectionException.class, action::run);
   }
 
-  private static void persistClaimsModeDomain() {
+  @Test
-    Domain domain = DatabaseHelper.newDomain("claims-landrush1.tld");
+  void testFailure_noLocationHeaderInResponse() throws Exception {
+    when(httpUrlConnection.getHeaderField(LOCATION)).thenReturn(null);
+    assertThrows(UrlConnectionException.class, action::run);
+  }
+
+  private void persistClaimsModeDomain() {
     persistDomainAndEnqueueLordn(
-        domain
+        newDomain("claims-landrush2.tld")
             .asBuilder()
+            .setCreationTimeForTest(clock.nowUtc())
+            .setCreationRegistrarId("NewRegistrar")
             .setLaunchNotice(
-                LaunchNotice.create(
+                LaunchNotice.create("landrush2tcn", null, null, clock.nowUtc().minusHours(2)))
-                    "landrush1tcn", null, null, domain.getCreationTime().minusHours(1)))
+            .setLordnPhase(LordnPhase.CLAIMS)
+            .build());
+    clock.advanceBy(Duration.standardDays(1));
+    persistDomainAndEnqueueLordn(
+        newDomain("claims-landrush1.tld")
+            .asBuilder()
+            .setCreationTimeForTest(clock.nowUtc())
+            .setLaunchNotice(
+                LaunchNotice.create("landrush1tcn", null, null, clock.nowUtc().minusHours(1)))
+            .setLordnPhase(LordnPhase.CLAIMS)
             .build());
   }
 
   private void persistSunriseModeDomain() {
-    action.phase = "sunrise";
+    persistDomainAndEnqueueLordn(
-    Domain domain = DatabaseHelper.newDomain("sunrise1.tld");
+        newDomain("sunrise2.tld")
-    persistDomainAndEnqueueLordn(domain.asBuilder().setSmdId("my-smdid").build());
+            .asBuilder()
+            .setCreationTimeForTest(clock.nowUtc())
+            .setCreationRegistrarId("NewRegistrar")
+            .setSmdId("new-smdid")
+            .setLordnPhase(LordnPhase.SUNRISE)
+            .build());
+    clock.advanceBy(Duration.standardDays(1));
+    persistDomainAndEnqueueLordn(
+        newDomain("sunrise1.tld")
+            .asBuilder()
+            .setCreationTimeForTest(clock.nowUtc())
+            .setSmdId("my-smdid")
+            .setLordnPhase(LordnPhase.SUNRISE)
+            .build());
   }
 
   private static TaskHandle makeTaskHandle(
@@ -311,4 +313,46 @@ class NordnUploadActionTest {
         TaskOptions.Builder.withPayload(payload).method(Method.PULL).tag(tag).taskName(taskName),
         queue);
   }
+
+  private void verifyColumnCleared(String domainName) {
+    VKey<Domain> domainKey = load(Domain.class, domainName, clock.nowUtc());
+    Domain domain = loadByKey(domainKey);
+    assertThat(domain.getLordnPhase()).isEqualTo(LordnPhase.NONE);
+  }
+
+  private void testRun(
+      String phase, String domain1, String domain2, String csv, boolean usePullQueue)
+      throws Exception {
+    action.usePullQueue = Optional.of(usePullQueue);
+    action.phase = phase;
+    action.run();
+    verify(httpUrlConnection)
+        .setRequestProperty(
+            AUTHORIZATION, "Basic bG9sY2F0OmF0dGFjaw=="); // echo -n lolcat:attack | base64
+    verify(httpUrlConnection)
+        .setRequestProperty(eq(CONTENT_TYPE), startsWith("multipart/form-data; boundary="));
+    verify(httpUrlConnection).setRequestMethod("POST");
+    assertThat(httpUrlConnection.getURL())
+        .isEqualTo(new URL("http://127.0.0.1/LORDN/tld/" + phase));
+    assertThat(connectionOutputStream.toString(UTF_8)).contains(csv);
+    if (!usePullQueue) {
+      verifyColumnCleared(domain1);
+      verifyColumnCleared(domain2);
+    } else {
+      assertThat(
+              getQueue("lordn-" + phase)
+                  .leaseTasks(
+                      LeaseOptions.Builder.withTag("tld")
+                          .leasePeriod(1, TimeUnit.HOURS)
+                          .countLimit(100)))
+          .isEmpty();
+    }
+    cloudTasksHelper.assertTasksEnqueued(
+        NordnVerifyAction.QUEUE,
+        new TaskMatcher()
+            .url(NordnVerifyAction.PATH)
+            .param(NordnVerifyAction.NORDN_URL_PARAM, LOCATION_URL)
+            .param(RequestParameters.PARAM_TLD, "tld")
+            .header(CONTENT_TYPE, FORM_DATA.toString()));
+  }
 }