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
-  private DatabaseMigrationStateSchedule() {}
+  // Required for Hibernate initialization
+  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),
-
-    CLAIMS(QUEUE_CLAIMS),
-
-    NONE(null);
-
-    final String queue;
-
-    LordnPhase(String queue) {
-      this.queue = queue;
-    }
+    SUNRISE,
+    CLAIMS,
+    NONE
   }
 }

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 @Parameter(LORDN_PHASE_PARAM) String phase;
-  @Inject @Parameter(RequestParameters.PARAM_TLD) String tld;
+  @Inject
+  @Config("tmchMarksdbUrl")
+  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 {
-      processLordnTasks();
-    } catch (IOException | GeneralSecurityException e) {
-      throw new RuntimeException(e);
+    if (usePullQueue.orElse(false)) {
+      try {
+        processLordnTasks();
+      } 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);