Make the verify OT&E action more performant

As previously written, it loaded up all history entries into memory and then
processed them. This was OOMing for some registrars on sandbox who had performed
a large number of testing actions, most of them long OT&E was passed.

This commit changes the verify OT&E action to stream the history entries in
batches, ordered by when they were made, and then terminates once all tests have
passed. This prevents OOMing because only a single batch of history entries need
reside in memory at once.

This does necessitate the creation of a new composite Datastore index on
HistoryEntry, so we'll need to run the ResaveAllHistoryEntriesAction in sandbox
after this change is deployed before the new verify OT&E code will work.

Note that the "history viewer" is long dead, but that the pre-existing index
on HistoryEntries is still used for many other purposes.

-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=223163337

java/google/registry/tools/server/VerifyOteAction.java

@@ -28,6 +28,8 @@ import com.google.common.base.Predicates;
 import com.google.common.collect.HashMultiset;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Multiset;
+import com.googlecode.objectify.Key;
+import com.googlecode.objectify.cmd.Query;
 import google.registry.flows.EppException;
 import google.registry.model.domain.DomainCommand;
 import google.registry.model.domain.fee.FeeCreateCommandExtension;
@@ -44,12 +46,14 @@ import google.registry.request.JsonActionRunner;
 import google.registry.request.JsonActionRunner.JsonAction;
 import google.registry.request.auth.Auth;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import java.util.stream.Stream;
 import javax.inject.Inject;
 
@@ -242,18 +246,30 @@ public class VerifyOteAction implements Runnable, JsonAction {
     /**
      * Records data in the passed historyEntryStats object on what actions have been performed by
      * the four numbered OT&E variants of the registrar name.
+     *
+     * <p>Stops when it notices that all tests have passed.
      */
     HistoryEntryStats recordRegistrarHistory(String registrarName) {
-      ImmutableList.Builder<String> clientIds = new ImmutableList.Builder<>();
-      for (int i = 1; i <= 4; i++) {
-        clientIds.add(String.format("%s-%d", registrarName, i));
-      }
-      for (HistoryEntry historyEntry :
-          ofy().load().type(HistoryEntry.class).filter("clientId in", clientIds.build()).list()) {
+      ImmutableList<String> clientIds =
+          IntStream.rangeClosed(1, 4)
+              .mapToObj(i -> String.format("%s-%d", registrarName, i))
+              .collect(toImmutableList());
+
+      Query<HistoryEntry> query =
+          ofy()
+              .load()
+              .type(HistoryEntry.class)
+              .filter("clientId in", clientIds)
+              .order("modificationTime");
+      for (HistoryEntry historyEntry : query) {
         try {
           record(historyEntry);
         } catch (EppException e) {
-          throw new RuntimeException(e);
+          throw new RuntimeException("Couldn't parse history entry " + Key.create(historyEntry), e);
+        }
+        // Break out early if all tests were passed.
+        if (wereAllTestsPassed()) {
+          break;
         }
       }
       return this;
@@ -264,18 +280,19 @@ public class VerifyOteAction implements Runnable, JsonAction {
       byte[] xmlBytes = historyEntry.getXmlBytes();
       // xmlBytes can be null on contact create and update for safe-harbor compliance.
       final Optional<EppInput> eppInput =
-          (xmlBytes == null)
-              ? Optional.empty()
-              : Optional.of(unmarshal(EppInput.class, xmlBytes));
+          (xmlBytes == null) ? Optional.empty() : Optional.of(unmarshal(EppInput.class, xmlBytes));
       if (!statCounts.addAll(
-          EnumSet.allOf(StatType.class)
-              .stream()
+          EnumSet.allOf(StatType.class).stream()
               .filter(statType -> statType.matches(historyEntry.getType(), eppInput))
               .collect(toImmutableList()))) {
         statCounts.add(StatType.UNCLASSIFIED_FLOWS);
       }
     }
 
+    boolean wereAllTestsPassed() {
+      return Arrays.stream(StatType.values()).allMatch(s -> statCounts.count(s) >= s.requirement);
+    }
+
     /**
      * Returns a list of failure messages describing any cases where the passed stats fail to meet
      * the required thresholds, or the empty list if all requirements are met.