Make ExpandRecurringBillingEventAction SQL-aware (#1181)

There is some complication regarding how the
CancellationMatchingBillingEvent of the generated OneTime can be
reconstructed when loading from SQL. I decided to only address it in
testing as there is no real value to fully reconstruct this VKey in
production where we are either in SQL or Ofy mode, both never in both.
Therefore the VKey in a particular mode only needs to contain the
corresponding key in order to function.

<!-- 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/1181)
<!-- Reviewable:end -->

core/src/main/java/google/registry/batch/ExpandRecurringBillingEventsAction.java

@@ -21,9 +21,12 @@ import static google.registry.mapreduce.MapreduceRunner.PARAM_DRY_RUN;
 import static google.registry.mapreduce.inputs.EppResourceInputs.createChildEntityInput;
 import static google.registry.model.common.Cursor.CursorType.RECURRING_BILLING;
 import static google.registry.model.domain.Period.Unit.YEARS;
-import static google.registry.model.ofy.ObjectifyService.ofy;
+import static google.registry.model.ofy.ObjectifyService.auditedOfy;
 import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_AUTORENEW;
+import static google.registry.persistence.transaction.QueryComposer.Comparator.EQ;
+import static google.registry.persistence.transaction.TransactionManagerFactory.jpaTm;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.persistence.transaction.TransactionManagerUtil.transactIfJpaTm;
 import static google.registry.pricing.PricingEngineProxy.getDomainRenewCost;
 import static google.registry.util.CollectionUtils.union;
 import static google.registry.util.DateTimeUtils.START_OF_TIME;
@@ -38,10 +41,8 @@ import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Range;
 import com.google.common.collect.Streams;
 import com.google.common.flogger.FluentLogger;
-import com.googlecode.objectify.Key;
 import google.registry.mapreduce.MapreduceRunner;
 import google.registry.mapreduce.inputs.NullInput;
-import google.registry.model.EppResource;
 import google.registry.model.ImmutableObject;
 import google.registry.model.billing.BillingEvent;
 import google.registry.model.billing.BillingEvent.Flag;
@@ -54,7 +55,7 @@ import google.registry.model.domain.Period;
 import google.registry.model.registry.Registry;
 import google.registry.model.reporting.DomainTransactionRecord;
 import google.registry.model.reporting.DomainTransactionRecord.TransactionReportField;
-import google.registry.model.reporting.HistoryEntry;
+import google.registry.persistence.VKey;
 import google.registry.request.Action;
 import google.registry.request.Parameter;
 import google.registry.request.Response;
@@ -92,17 +93,20 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
 
   @Override
   public void run() {
-    Cursor cursor =
-        tm().loadByKeyIfPresent(Cursor.createGlobalVKey(RECURRING_BILLING)).orElse(null);
     DateTime executeTime = clock.nowUtc();
-    DateTime persistedCursorTime = (cursor == null ? START_OF_TIME : cursor.getCursorTime());
+    DateTime persistedCursorTime =
+        transactIfJpaTm(
+            () ->
+                tm().loadByKeyIfPresent(Cursor.createGlobalVKey(RECURRING_BILLING))
+                    .orElse(Cursor.createGlobal(RECURRING_BILLING, START_OF_TIME))
+                    .getCursorTime());
     DateTime cursorTime = cursorTimeParam.orElse(persistedCursorTime);
     checkArgument(
-        cursorTime.isBefore(executeTime),
+        cursorTime.isBefore(executeTime), "Cursor time must be earlier than execution time.");
-        "Cursor time must be earlier than execution time.");
     logger.atInfo().log(
         "Running Recurring billing event expansion for billing time range [%s, %s).",
         cursorTime, executeTime);
+    if (tm().isOfy()) {
       mrRunner
           .setJobName("Expand Recurring billing events into synthetic OneTime events.")
           .setModuleName("backend")
@@ -115,8 +119,61 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
                   createChildEntityInput(
                       ImmutableSet.of(DomainBase.class), ImmutableSet.of(Recurring.class))))
           .sendLinkToMapreduceConsole(response);
-  }
+    } else {
+      int numBillingEventsSaved =
+          jpaTm()
+              .transact(
+                  () ->
+                      jpaTm()
+                          .query(
+                              "FROM BillingRecurrence "
+                                  + "WHERE event_time <= :executeTime "
+                                  + "AND event_time < recurrence_end_time",
+                              Recurring.class)
+                          .setParameter("executeTime", executeTime.toDate())
+                          // Need to get a list from the transaction and then convert it to a stream
+                          // for further processing. If we get a stream directly, each elements gets
+                          // processed downstream eagerly but Hibernate returns a
+                          // ScrollableResultsIterator that cannot be advanced outside the
+                          // transaction, resulting in an exception.
+                          .getResultList())
+              .stream()
+              .map(
+                  recurring ->
+                      jpaTm()
+                          .transact(
+                              () ->
+                                  expandBillingEvent(recurring, executeTime, cursorTime, isDryRun)))
+              .reduce(0, Integer::sum);
 
+      if (!isDryRun) {
+        logger.atInfo().log("Saved OneTime billing events", numBillingEventsSaved);
+      } else {
+        logger.atInfo().log("Generated OneTime billing events (dry run)", numBillingEventsSaved);
+      }
+      logger.atInfo().log(
+          "Recurring event expansion %s complete for billing event range [%s, %s).",
+          isDryRun ? "(dry run) " : "", cursorTime, executeTime);
+      tm().transact(
+              () -> {
+                // Check for the unlikely scenario where the cursor has been altered during the
+                // expansion.
+                DateTime currentCursorTime =
+                    tm().loadByKeyIfPresent(Cursor.createGlobalVKey(RECURRING_BILLING))
+                        .orElse(Cursor.createGlobal(RECURRING_BILLING, START_OF_TIME))
+                        .getCursorTime();
+                if (!currentCursorTime.equals(persistedCursorTime)) {
+                  throw new IllegalStateException(
+                      String.format(
+                          "Current cursor position %s does not match persisted cursor position %s.",
+                          currentCursorTime, persistedCursorTime));
+                }
+                if (!isDryRun) {
+                  tm().put(Cursor.createGlobal(RECURRING_BILLING, executeTime));
+                }
+              });
+    }
+  }
   /** Mapper to expand {@link Recurring} billing events into synthetic {@link OneTime} events. */
   public static class ExpandRecurringBillingEventsMapper
       extends Mapper<Recurring, DateTime, DateTime> {
@@ -155,11 +212,77 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
       try {
         numBillingEventsSaved =
             tm().transactNew(
+                    () -> expandBillingEvent(recurring, executeTime, cursorTime, isDryRun));
+      } catch (Throwable t) {
+        getContext().incrementCounter("error: " + t.getClass().getSimpleName());
+        getContext().incrementCounter(ERROR_COUNTER);
+        throw new RuntimeException(
+            String.format(
+                "Error while expanding Recurring billing events for %d", recurring.getId()),
+            t);
+      }
+      if (!isDryRun) {
+        getContext().incrementCounter("Saved OneTime billing events", numBillingEventsSaved);
+      } else {
+        getContext()
+            .incrementCounter("Generated OneTime billing events (dry run)", numBillingEventsSaved);
+      }
+    }
+  }
+
+  /**
+   * "Reducer" to advance the cursor after all map jobs have been completed. The NullInput into the
+   * mapper will cause the mapper to emit one timestamp pair (current cursor and execution time),
+   * and the cursor will be advanced (and the timestamps logged) at the end of a successful
+   * mapreduce.
+   */
+  public static class ExpandRecurringBillingEventsReducer
+      extends Reducer<DateTime, DateTime, Void> {
+
+    private final boolean isDryRun;
+    private final DateTime expectedPersistedCursorTime;
+
+    public ExpandRecurringBillingEventsReducer(
+        boolean isDryRun, DateTime expectedPersistedCursorTime) {
+      this.isDryRun = isDryRun;
+      this.expectedPersistedCursorTime = expectedPersistedCursorTime;
+    }
+
+    @Override
+    public void reduce(final DateTime cursorTime, final ReducerInput<DateTime> executionTimeInput) {
+      if (getContext().getCounter(ERROR_COUNTER).getValue() > 0) {
+        logger.atSevere().log(
+            "One or more errors logged during recurring event expansion. Cursor will"
+                + " not be advanced.");
+        return;
+      }
+      final DateTime executionTime = executionTimeInput.next();
+      logger.atInfo().log(
+          "Recurring event expansion %s complete for billing event range [%s, %s).",
+          isDryRun ? "(dry run) " : "", cursorTime, executionTime);
+      tm().transact(
               () -> {
-                      ImmutableSet.Builder<OneTime> syntheticOneTimesBuilder =
+                Cursor cursor =
-                          new ImmutableSet.Builder<>();
+                    auditedOfy().load().key(Cursor.createGlobalKey(RECURRING_BILLING)).now();
-                      final Registry tld =
+                DateTime currentCursorTime =
-                          Registry.get(getTldFromDomainName(recurring.getTargetId()));
+                    (cursor == null ? START_OF_TIME : cursor.getCursorTime());
+                if (!currentCursorTime.equals(expectedPersistedCursorTime)) {
+                  logger.atSevere().log(
+                      "Current cursor position %s does not match expected cursor position %s.",
+                      currentCursorTime, expectedPersistedCursorTime);
+                  return;
+                }
+                if (!isDryRun) {
+                  tm().put(Cursor.createGlobal(RECURRING_BILLING, executionTime));
+                }
+              });
+    }
+  }
+
+  private static int expandBillingEvent(
+      Recurring recurring, DateTime executeTime, DateTime cursorTime, boolean isDryRun) {
+    ImmutableSet.Builder<OneTime> syntheticOneTimesBuilder = new ImmutableSet.Builder<>();
+    final Registry tld = Registry.get(getTldFromDomainName(recurring.getTargetId()));
 
     // Determine the complete set of times at which this recurring event should
     // occur (up to and including the runtime of the mapreduce).
@@ -175,16 +298,24 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
     final ImmutableSet<DateTime> billingTimes =
         getBillingTimesInScope(eventTimes, cursorTime, executeTime, tld);
 
-                      Key<? extends EppResource> domainKey = recurring.getParentKey().getParent();
+    VKey<DomainBase> domainKey =
-                      Iterable<OneTime> oneTimesForDomain =
+        VKey.create(
-                          ofy().load().type(OneTime.class).ancestor(domainKey);
+            DomainBase.class, recurring.getDomainRepoId(), recurring.getParentKey().getParent());
+    Iterable<OneTime> oneTimesForDomain;
+    if (tm().isOfy()) {
+      oneTimesForDomain = auditedOfy().load().type(OneTime.class).ancestor(domainKey.getOfyKey());
+    } else {
+      oneTimesForDomain =
+          tm().createQueryComposer(OneTime.class)
+              .where("domainRepoId", EQ, recurring.getDomainRepoId())
+              .list();
+    }
 
     // Determine the billing times that already have OneTime events persisted.
     ImmutableSet<DateTime> existingBillingTimes =
         getExistingBillingTimes(oneTimesForDomain, recurring);
 
-                      ImmutableSet.Builder<HistoryEntry> historyEntriesBuilder =
+    ImmutableSet.Builder<DomainHistory> historyEntriesBuilder = new ImmutableSet.Builder<>();
-                          new ImmutableSet.Builder<>();
     // Create synthetic OneTime events for all billing times that do not yet have
     // an event persisted.
     for (DateTime billingTime : difference(billingTimes, existingBillingTimes)) {
@@ -194,12 +325,9 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
               .setBySuperuser(false)
               .setClientId(recurring.getClientId())
               .setModificationTime(tm().getTransactionTime())
-                                // TODO (jianglai): modify this to use setDomain instead when
+              .setDomain(tm().loadByKey(domainKey))
-                                //  converting this action to be SQL-aware.
-                                .setDomainRepoId(domainKey.getName())
               .setPeriod(Period.create(1, YEARS))
-                                .setReason(
+              .setReason("Domain autorenewal by ExpandRecurringBillingEventsAction")
-                                    "Domain autorenewal by ExpandRecurringBillingEventsAction")
               .setRequestedByRegistrar(false)
               .setType(DOMAIN_AUTORENEW)
               // Don't write a domain transaction record if the recurrence was
@@ -237,7 +365,7 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
               .setTargetId(recurring.getTargetId())
               .build());
     }
-                      Set<HistoryEntry> historyEntries = historyEntriesBuilder.build();
+    Set<DomainHistory> historyEntries = historyEntriesBuilder.build();
     Set<OneTime> syntheticOneTimes = syntheticOneTimesBuilder.build();
     if (!isDryRun) {
       ImmutableSet<ImmutableObject> entitiesToSave =
@@ -245,31 +373,16 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
               .addAll(historyEntries)
               .addAll(syntheticOneTimes)
               .build();
-                        ofy().save().entities(entitiesToSave).now();
+      tm().putAll(entitiesToSave);
     }
     return syntheticOneTimes.size();
-                    });
-      } catch (Throwable t) {
-        getContext().incrementCounter("error: " + t.getClass().getSimpleName());
-        getContext().incrementCounter(ERROR_COUNTER);
-        throw new RuntimeException(
-            String.format(
-                "Error while expanding Recurring billing events for %d", recurring.getId()),
-            t);
-      }
-      if (!isDryRun) {
-        getContext().incrementCounter("Saved OneTime billing events", numBillingEventsSaved);
-      } else {
-        getContext().incrementCounter(
-            "Generated OneTime billing events (dry run)", numBillingEventsSaved);
-      }
   }
 
   /**
    * Filters a set of {@link DateTime}s down to event times that are in scope for a particular
    * mapreduce run, given the cursor time and the mapreduce execution time.
    */
-    private ImmutableSet<DateTime> getBillingTimesInScope(
+  protected static ImmutableSet<DateTime> getBillingTimesInScope(
       Iterable<DateTime> eventTimes,
       DateTime cursorTime,
       DateTime executeTime,
@@ -281,10 +394,10 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
   }
 
   /**
-     * Determines an {@link ImmutableSet} of {@link DateTime}s that have already been persisted
+   * Determines an {@link ImmutableSet} of {@link DateTime}s that have already been persisted for a
-     * for a given recurring billing event.
+   * given recurring billing event.
    */
-    private ImmutableSet<DateTime> getExistingBillingTimes(
+  private static ImmutableSet<DateTime> getExistingBillingTimes(
       Iterable<BillingEvent.OneTime> oneTimesForDomain,
       final BillingEvent.Recurring recurringEvent) {
     return Streams.stream(oneTimesForDomain)
@@ -296,54 +409,4 @@ public class ExpandRecurringBillingEventsAction implements Runnable {
         .map(OneTime::getBillingTime)
         .collect(toImmutableSet());
   }
-  }
-
-
-  /**
-   * "Reducer" to advance the cursor after all map jobs have been completed. The NullInput into the
-   * mapper will cause the mapper to emit one timestamp pair (current cursor and execution time),
-   * and the cursor will be advanced (and the timestamps logged) at the end of a successful
-   * mapreduce.
-   */
-  public static class ExpandRecurringBillingEventsReducer
-      extends Reducer<DateTime, DateTime, Void> {
-
-    private final boolean isDryRun;
-    private final DateTime expectedPersistedCursorTime;
-
-    public ExpandRecurringBillingEventsReducer(
-        boolean isDryRun, DateTime expectedPersistedCursorTime) {
-      this.isDryRun = isDryRun;
-      this.expectedPersistedCursorTime = expectedPersistedCursorTime;
-    }
-
-    @Override
-    public void reduce(final DateTime cursorTime, final ReducerInput<DateTime> executionTimeInput) {
-      if (getContext().getCounter(ERROR_COUNTER).getValue() > 0) {
-        logger.atSevere().log(
-            "One or more errors logged during recurring event expansion. Cursor will"
-                + " not be advanced.");
-        return;
-      }
-      final DateTime executionTime = executionTimeInput.next();
-      logger.atInfo().log(
-          "Recurring event expansion %s complete for billing event range [%s, %s).",
-          isDryRun ? "(dry run) " : "", cursorTime, executionTime);
-      tm().transact(
-              () -> {
-                Cursor cursor = ofy().load().key(Cursor.createGlobalKey(RECURRING_BILLING)).now();
-                DateTime currentCursorTime =
-                    (cursor == null ? START_OF_TIME : cursor.getCursorTime());
-                if (!currentCursorTime.equals(expectedPersistedCursorTime)) {
-                  logger.atSevere().log(
-                      "Current cursor position %s does not match expected cursor position %s.",
-                      currentCursorTime, expectedPersistedCursorTime);
-                  return;
-                }
-                if (!isDryRun) {
-                  tm().put(Cursor.createGlobal(RECURRING_BILLING, executionTime));
-                }
-              });
-    }
-  }
 }

core/src/main/java/google/registry/model/billing/BillingEvent.java

@@ -324,13 +324,22 @@ public abstract class BillingEvent extends ImmutableObject
     DateTime syntheticCreationTime;
 
     /**
-     * For {@link Flag#SYNTHETIC} events, a {@link Key} to the {@link BillingEvent} from which this
+     * For {@link Flag#SYNTHETIC} events, a {@link Key} to the {@link Recurring} from which this
-     * OneTime was created. This is needed in order to properly match billing events against {@link
+     * {@link OneTime} was created. This is needed in order to properly match billing events against
-     * Cancellation}s.
+     * {@link Cancellation}s.
      */
     @Column(name = "cancellation_matching_billing_recurrence_id")
     VKey<Recurring> cancellationMatchingBillingEvent;
 
+    /**
+     * For {@link Flag#SYNTHETIC} events, the {@link DomainHistory} revision ID of the the {@link
+     * Recurring} from which this {@link OneTime} was created. This is needed in order to recreate
+     * the {@link VKey} when reading from SQL.
+     */
+    @Ignore
+    @Column(name = "recurrence_history_revision_id")
+    Long recurringEventHistoryRevisionId;
+
     /**
      * The {@link AllocationToken} used in the creation of this event, or null if one was not used.
      */
@@ -354,10 +363,14 @@ public abstract class BillingEvent extends ImmutableObject
       return syntheticCreationTime;
     }
 
-    public VKey<? extends BillingEvent> getCancellationMatchingBillingEvent() {
+    public VKey<Recurring> getCancellationMatchingBillingEvent() {
       return cancellationMatchingBillingEvent;
     }
 
+    public Long getRecurringEventHistoryRevisionId() {
+      return recurringEventHistoryRevisionId;
+    }
+
     public Optional<VKey<AllocationToken>> getAllocationToken() {
       return Optional.ofNullable(allocationToken);
     }
@@ -376,6 +389,28 @@ public abstract class BillingEvent extends ImmutableObject
       return new Builder(clone(this));
     }
 
+    @Override
+    void onLoad() {
+      super.onLoad();
+      if (cancellationMatchingBillingEvent != null) {
+        recurringEventHistoryRevisionId =
+            cancellationMatchingBillingEvent.getOfyKey().getParent().getId();
+      }
+    }
+
+    @Override
+    void postLoad() {
+      super.postLoad();
+      if (cancellationMatchingBillingEvent != null) {
+        cancellationMatchingBillingEvent =
+            cancellationMatchingBillingEvent.restoreOfy(
+                DomainBase.class,
+                domainRepoId,
+                DomainHistory.class,
+                recurringEventHistoryRevisionId);
+      }
+    }
+
     /** A builder for {@link OneTime} since it is immutable. */
     public static class Builder extends BillingEvent.Builder<OneTime, Builder> {
 
@@ -410,6 +445,8 @@ public abstract class BillingEvent extends ImmutableObject
       public Builder setCancellationMatchingBillingEvent(
           VKey<Recurring> cancellationMatchingBillingEvent) {
         getInstance().cancellationMatchingBillingEvent = cancellationMatchingBillingEvent;
+        getInstance().recurringEventHistoryRevisionId =
+            cancellationMatchingBillingEvent.getOfyKey().getParent().getId();
         return this;
       }
 
@@ -444,6 +481,11 @@ public abstract class BillingEvent extends ImmutableObject
                 == (instance.cancellationMatchingBillingEvent != null),
             "Cancellation matching billing event must be set if and only if the SYNTHETIC flag "
                 + "is set.");
+        checkState(
+            !instance.getFlags().contains(Flag.SYNTHETIC)
+                || (instance.cancellationMatchingBillingEvent.getOfyKey().getParent().getId()
+                    == instance.recurringEventHistoryRevisionId),
+            "Cancellation matching billing event and its history revision ID does not match.");
         return super.build();
       }
     }

core/src/main/java/google/registry/model/reporting/HistoryEntryDao.java

@@ -184,6 +184,7 @@ public class HistoryEntryDao {
             .where("modificationTime", criteriaBuilder::greaterThanOrEqualTo, afterTime)
             .where("modificationTime", criteriaBuilder::lessThanOrEqualTo, beforeTime)
             .where(repoIdFieldName, criteriaBuilder::equal, parentKey.getSqlKey().toString())
+            .orderByAsc("id")
             .build();
 
     return ImmutableList.sortedCopyOf(

core/src/test/java/google/registry/batch/ExpandRecurringBillingEventsActionTest.java

@@ -21,6 +21,7 @@ import static google.registry.model.ofy.ObjectifyService.auditedOfy;
 import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_AUTORENEW;
 import static google.registry.model.reporting.HistoryEntry.Type.DOMAIN_CREATE;
 import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.persistence.transaction.TransactionManagerUtil.transactIfJpaTm;
 import static google.registry.testing.DatabaseHelper.assertBillingEvents;
 import static google.registry.testing.DatabaseHelper.assertBillingEventsForResource;
 import static google.registry.testing.DatabaseHelper.createTld;
@@ -52,9 +53,14 @@ import google.registry.model.registry.Registry;
 import google.registry.model.reporting.DomainTransactionRecord;
 import google.registry.model.reporting.DomainTransactionRecord.TransactionReportField;
 import google.registry.model.reporting.HistoryEntry;
+import google.registry.testing.DualDatabaseTest;
 import google.registry.testing.FakeClock;
 import google.registry.testing.FakeResponse;
 import google.registry.testing.InjectExtension;
+import google.registry.testing.ReplayExtension;
+import google.registry.testing.TestOfyAndSql;
+import google.registry.testing.TestOfyOnly;
+import google.registry.testing.TestSqlOnly;
 import google.registry.testing.mapreduce.MapreduceTestCase;
 import java.util.ArrayList;
 import java.util.List;
@@ -62,17 +68,25 @@ import java.util.Optional;
 import org.joda.money.Money;
 import org.joda.time.DateTime;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Order;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
 /** Unit tests for {@link ExpandRecurringBillingEventsAction}. */
+@DualDatabaseTest
 public class ExpandRecurringBillingEventsActionTest
     extends MapreduceTestCase<ExpandRecurringBillingEventsAction> {
 
-  @RegisterExtension public final InjectExtension inject = new InjectExtension();
+  private DateTime currentTestTime = DateTime.parse("1999-01-05T00:00:00Z");
+  private final FakeClock clock = new FakeClock(currentTestTime);
 
-  private final DateTime beginningOfTest = DateTime.parse("2000-10-02T00:00:00Z");
+  @Order(Order.DEFAULT - 1)
-  private final FakeClock clock = new FakeClock(beginningOfTest);
+  @RegisterExtension
+  public final InjectExtension inject =
+      new InjectExtension().withStaticFieldOverride(Ofy.class, "clock", clock);
+
+  @Order(Order.DEFAULT - 2)
+  @RegisterExtension
+  public final ReplayExtension replayExtension = ReplayExtension.createWithCompare(clock);
 
   private DomainBase domain;
   private DomainHistory historyEntry;
@@ -80,7 +94,6 @@ public class ExpandRecurringBillingEventsActionTest
 
   @BeforeEach
   void beforeEach() {
-    inject.setStaticField(Ofy.class, "clock", clock);
     action = new ExpandRecurringBillingEventsAction();
     action.mrRunner = makeDefaultRunner();
     action.clock = clock;
@@ -111,27 +124,37 @@ public class ExpandRecurringBillingEventsActionTest
             .setRecurrenceEndTime(END_OF_TIME)
             .setTargetId(domain.getDomainName())
             .build();
+    currentTestTime = clock.nowUtc();
+    clock.setTo(DateTime.parse("2000-10-02T00:00:00Z"));
   }
 
   private void saveCursor(final DateTime cursorTime) {
     tm().transact(() -> tm().put(Cursor.createGlobal(RECURRING_BILLING, cursorTime)));
   }
 
-  private void runMapreduce() throws Exception {
+  private void runAction() throws Exception {
     action.response = new FakeResponse();
     action.run();
+    // Need to save the current test time before running the mapreduce, which increments the clock.
+    // The execution time (e. g. transaction time) is captured when the action starts running so
+    // the passage of time afterward does not affect the timestamp stored in the billing events.
+    currentTestTime = clock.nowUtc();
     executeTasksUntilEmpty("mapreduce", clock);
     auditedOfy().clearSessionCache();
   }
 
   private void assertCursorAt(DateTime expectedCursorTime) {
-    Cursor cursor = auditedOfy().load().key(Cursor.createGlobalKey(RECURRING_BILLING)).now();
+    Cursor cursor =
+        transactIfJpaTm(() -> tm().loadByKey(Cursor.createGlobalVKey(RECURRING_BILLING)));
     assertThat(cursor).isNotNull();
     assertThat(cursor.getCursorTime()).isEqualTo(expectedCursorTime);
   }
 
   private void assertHistoryEntryMatches(
-      DomainBase domain, HistoryEntry actual, String clientId, DateTime billingTime,
+      DomainBase domain,
+      HistoryEntry actual,
+      String clientId,
+      DateTime billingTime,
       boolean shouldHaveTxRecord) {
     assertThat(actual.getBySuperuser()).isFalse();
     assertThat(actual.getClientId()).isEqualTo(clientId);
@@ -145,10 +168,7 @@ public class ExpandRecurringBillingEventsActionTest
       assertThat(actual.getDomainTransactionRecords())
           .containsExactly(
               DomainTransactionRecord.create(
-                  "tld",
+                  "tld", billingTime, TransactionReportField.NET_RENEWS_1_YR, 1));
-                  billingTime,
-                  TransactionReportField.NET_RENEWS_1_YR,
-                  1));
     } else {
       assertThat(actual.getDomainTransactionRecords()).isEmpty();
     }
@@ -163,28 +183,26 @@ public class ExpandRecurringBillingEventsActionTest
         .setFlags(ImmutableSet.of(Flag.AUTO_RENEW, Flag.SYNTHETIC))
         .setPeriodYears(1)
         .setReason(Reason.RENEW)
-        .setSyntheticCreationTime(beginningOfTest)
+        .setSyntheticCreationTime(currentTestTime)
         .setCancellationMatchingBillingEvent(recurring.createVKey())
         .setTargetId(domain.getDomainName());
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent() throws Exception {
     persistResource(recurring);
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     DomainHistory persistedEntry =
         getOnlyHistoryEntryOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertHistoryEntryMatches(
         domain, persistedEntry, "TheRegistrar", DateTime.parse("2000-02-19T00:00:00Z"), true);
-    BillingEvent.OneTime expected = defaultOneTimeBuilder()
+    BillingEvent.OneTime expected = defaultOneTimeBuilder().setParent(persistedEntry).build();
-        .setParent(persistedEntry)
-        .build();
     assertBillingEventsForResource(domain, expected, recurring);
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_deletedDomain() throws Exception {
     DateTime deletionTime = DateTime.parse("2000-08-01T00:00:00Z");
     DomainBase deletedDomain = persistDeletedDomain("deleted.tld", deletionTime);
@@ -209,7 +227,7 @@ public class ExpandRecurringBillingEventsActionTest
                 .setTargetId(deletedDomain.getDomainName())
                 .build());
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     DomainHistory persistedEntry =
         getOnlyHistoryEntryOfType(deletedDomain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertHistoryEntryMatches(
@@ -224,69 +242,69 @@ public class ExpandRecurringBillingEventsActionTest
             .setTargetId(deletedDomain.getDomainName())
             .build();
     assertBillingEventsForResource(deletedDomain, expected, recurring);
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_idempotentForDuplicateRuns() throws Exception {
     persistResource(recurring);
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     DomainHistory persistedEntry =
         getOnlyHistoryEntryOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertHistoryEntryMatches(
         domain, persistedEntry, "TheRegistrar", DateTime.parse("2000-02-19T00:00:00Z"), true);
     BillingEvent.OneTime expected = defaultOneTimeBuilder().setParent(persistedEntry).build();
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
     DateTime beginningOfSecondRun = clock.nowUtc();
     action.response = new FakeResponse();
-    runMapreduce();
+    runAction();
     assertCursorAt(beginningOfSecondRun);
     assertBillingEventsForResource(domain, expected, recurring);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_idempotentForExistingOneTime() throws Exception {
     persistResource(recurring);
-    BillingEvent.OneTime persisted = persistResource(defaultOneTimeBuilder()
+    BillingEvent.OneTime persisted =
-        .setParent(historyEntry)
+        persistResource(defaultOneTimeBuilder().setParent(historyEntry).build());
-        .build());
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     // No new history entries should be generated
     assertThat(getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW)).isEmpty();
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
     // No additional billing events should be generated
     assertBillingEventsForResource(domain, persisted, recurring);
   }
 
-  @Test
+  @TestSqlOnly
   void testSuccess_expandSingleEvent_notIdempotentForDifferentBillingTime() throws Exception {
     persistResource(recurring);
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     DomainHistory persistedEntry =
         getOnlyHistoryEntryOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertHistoryEntryMatches(
         domain, persistedEntry, "TheRegistrar", DateTime.parse("2000-02-19T00:00:00Z"), true);
     BillingEvent.OneTime expected = defaultOneTimeBuilder().setParent(persistedEntry).build();
     // Persist an otherwise identical billing event that differs only in billing time.
-    BillingEvent.OneTime persisted = persistResource(expected.asBuilder()
+    BillingEvent.OneTime persisted =
+        persistResource(
+            expected
+                .asBuilder()
                 .setBillingTime(DateTime.parse("1999-02-19T00:00:00Z"))
                 .setEventTime(DateTime.parse("1999-01-05T00:00:00Z"))
                 .build());
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
     assertBillingEventsForResource(domain, persisted, expected, recurring);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_notIdempotentForDifferentRecurring() throws Exception {
     persistResource(recurring);
-    BillingEvent.Recurring recurring2 = persistResource(recurring.asBuilder()
+    BillingEvent.Recurring recurring2 = persistResource(recurring.asBuilder().setId(3L).build());
-        .setId(3L)
-        .build());
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     List<DomainHistory> persistedEntries =
         getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     for (HistoryEntry persistedEntry : persistedEntries) {
@@ -294,9 +312,8 @@ public class ExpandRecurringBillingEventsActionTest
           domain, persistedEntry, "TheRegistrar", DateTime.parse("2000-02-19T00:00:00Z"), true);
     }
     assertThat(persistedEntries).hasSize(2);
-    BillingEvent.OneTime expected = defaultOneTimeBuilder()
+    BillingEvent.OneTime expected =
-        .setParent(persistedEntries.get(0))
+        defaultOneTimeBuilder().setParent(persistedEntries.get(0)).build();
-        .build();
     // Persist an otherwise identical billing event that differs only in recurring event key.
     BillingEvent.OneTime persisted =
         expected
@@ -304,164 +321,166 @@ public class ExpandRecurringBillingEventsActionTest
             .setParent(persistedEntries.get(1))
             .setCancellationMatchingBillingEvent(recurring2.createVKey())
             .build();
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
     assertBillingEventsForResource(domain, persisted, expected, recurring, recurring2);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_ignoreRecurringBeforeWindow() throws Exception {
-    recurring = persistResource(recurring.asBuilder()
+    recurring =
+        persistResource(
+            recurring
+                .asBuilder()
                 .setEventTime(DateTime.parse("1997-01-05T00:00:00Z"))
                 .setRecurrenceEndTime(DateTime.parse("1999-10-05T00:00:00Z"))
                 .build());
     action.cursorTimeParam = Optional.of(DateTime.parse("2000-01-01T00:00:00Z"));
-    runMapreduce();
+    runAction();
     // No new history entries should be generated
     assertThat(getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW)).isEmpty();
     assertBillingEventsForResource(domain, recurring);
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_ignoreRecurringAfterWindow() throws Exception {
-    recurring = persistResource(recurring.asBuilder()
+    recurring =
-        .setEventTime(clock.nowUtc().plusYears(2))
+        persistResource(recurring.asBuilder().setEventTime(clock.nowUtc().plusYears(2)).build());
-        .build());
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     // No new history entries should be generated
     assertThat(getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW)).isEmpty();
     assertBillingEventsForResource(domain, recurring);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_billingTimeAtCursorTime() throws Exception {
     persistResource(recurring);
     action.cursorTimeParam = Optional.of(DateTime.parse("2000-02-19T00:00:00Z"));
-    runMapreduce();
+    runAction();
     DomainHistory persistedEntry =
         getOnlyHistoryEntryOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertHistoryEntryMatches(
         domain, persistedEntry, "TheRegistrar", DateTime.parse("2000-02-19T00:00:00Z"), true);
     BillingEvent.OneTime expected = defaultOneTimeBuilder().setParent(persistedEntry).build();
     assertBillingEventsForResource(domain, expected, recurring);
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_cursorTimeBetweenEventAndBillingTime() throws Exception {
     persistResource(recurring);
     action.cursorTimeParam = Optional.of(DateTime.parse("2000-01-12T00:00:00Z"));
-    runMapreduce();
+    runAction();
     DomainHistory persistedEntry =
         getOnlyHistoryEntryOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertHistoryEntryMatches(
         domain, persistedEntry, "TheRegistrar", DateTime.parse("2000-02-19T00:00:00Z"), true);
     BillingEvent.OneTime expected = defaultOneTimeBuilder().setParent(persistedEntry).build();
     assertBillingEventsForResource(domain, expected, recurring);
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_billingTimeAtExecutionTime() throws Exception {
-    DateTime testTime = DateTime.parse("2000-02-19T00:00:00Z").minusMillis(1);
+    clock.setTo(currentTestTime);
     persistResource(recurring);
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    // Clock is advanced one milli in runMapreduce()
+    clock.setTo(DateTime.parse("2000-02-19T00:00:00Z"));
-    clock.setTo(testTime);
+    runAction();
-    runMapreduce();
     // No new history entries should be generated
     assertThat(getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW)).isEmpty();
     // A candidate billing event is set to be billed exactly on 2/19/00 @ 00:00,
     // but these should not be generated as the interval is closed on cursorTime, open on
     // executeTime.
     assertBillingEventsForResource(domain, recurring);
-    assertCursorAt(testTime);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_multipleYearCreate() throws Exception {
-    DateTime testTime = beginningOfTest.plusYears(2);
     action.cursorTimeParam = Optional.of(recurring.getEventTime());
     recurring =
         persistResource(
             recurring.asBuilder().setEventTime(recurring.getEventTime().plusYears(2)).build());
-    clock.setTo(testTime);
+    clock.setTo(DateTime.parse("2002-10-02T00:00:00Z"));
-    runMapreduce();
+    runAction();
     DomainHistory persistedEntry =
         getOnlyHistoryEntryOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertHistoryEntryMatches(
         domain, persistedEntry, "TheRegistrar", DateTime.parse("2002-02-19T00:00:00Z"), true);
-    BillingEvent.OneTime expected = defaultOneTimeBuilder()
+    BillingEvent.OneTime expected =
+        defaultOneTimeBuilder()
             .setBillingTime(DateTime.parse("2002-02-19T00:00:00Z"))
             .setEventTime(DateTime.parse("2002-01-05T00:00:00Z"))
             .setParent(persistedEntry)
-        .setSyntheticCreationTime(testTime)
             .build();
     assertBillingEventsForResource(domain, expected, recurring);
-    assertCursorAt(testTime);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_withCursor() throws Exception {
     persistResource(recurring);
     saveCursor(START_OF_TIME);
-    runMapreduce();
+    runAction();
     DomainHistory persistedEntry =
         getOnlyHistoryEntryOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertHistoryEntryMatches(
         domain, persistedEntry, "TheRegistrar", DateTime.parse("2000-02-19T00:00:00Z"), true);
     BillingEvent.OneTime expected = defaultOneTimeBuilder().setParent(persistedEntry).build();
     assertBillingEventsForResource(domain, expected, recurring);
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_withCursorPastExpected() throws Exception {
     persistResource(recurring);
     // Simulate a quick second run of the mapreduce (this should be a no-op).
     saveCursor(clock.nowUtc().minusSeconds(1));
-    runMapreduce();
+    runAction();
     // No new history entries should be generated
     assertThat(getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW)).isEmpty();
     assertBillingEventsForResource(domain, recurring);
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_recurrenceEndBeforeEvent() throws Exception {
     // This can occur when a domain is transferred or deleted before a domain comes up for renewal.
-    recurring = persistResource(recurring.asBuilder()
+    recurring =
+        persistResource(
+            recurring
+                .asBuilder()
                 .setRecurrenceEndTime(recurring.getEventTime().minusDays(5))
                 .build());
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     // No new history entries should be generated
     assertThat(getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW)).isEmpty();
     assertBillingEventsForResource(domain, recurring);
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_dryRun() throws Exception {
     persistResource(recurring);
     action.isDryRun = true;
     saveCursor(START_OF_TIME); // Need a saved cursor to verify that it didn't move.
-    runMapreduce();
+    runAction();
     // No new history entries should be generated
     assertThat(getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW)).isEmpty();
     assertBillingEventsForResource(domain, recurring);
     assertCursorAt(START_OF_TIME); // Cursor doesn't move on a dry run.
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_multipleYears() throws Exception {
-    DateTime testTime = clock.nowUtc().plusYears(5);
+    clock.setTo(clock.nowUtc().plusYears(5));
-    clock.setTo(testTime);
     List<BillingEvent> expectedEvents = new ArrayList<>();
     expectedEvents.add(persistResource(recurring));
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     List<DomainHistory> persistedEntries =
         getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertThat(persistedEntries).hasSize(6);
@@ -470,29 +489,25 @@ public class ExpandRecurringBillingEventsActionTest
     // Expecting events for '00, '01, '02, '03, '04, '05.
     for (int year = 0; year < 6; year++) {
       assertHistoryEntryMatches(
-          domain,
+          domain, persistedEntries.get(year), "TheRegistrar", billingDate.plusYears(year), true);
-          persistedEntries.get(year),
+      expectedEvents.add(
-          "TheRegistrar",
+          defaultOneTimeBuilder()
-          billingDate.plusYears(year), true);
-      expectedEvents.add(defaultOneTimeBuilder()
               .setBillingTime(billingDate.plusYears(year))
               .setEventTime(eventDate.plusYears(year))
               .setParent(persistedEntries.get(year))
-          .setSyntheticCreationTime(testTime)
               .build());
     }
     assertBillingEventsForResource(domain, Iterables.toArray(expectedEvents, BillingEvent.class));
-    assertCursorAt(testTime);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_multipleYears_cursorInBetweenYears() throws Exception {
-    DateTime testTime = clock.nowUtc().plusYears(5);
+    clock.setTo(clock.nowUtc().plusYears(5));
-    clock.setTo(testTime);
     List<BillingEvent> expectedEvents = new ArrayList<>();
     expectedEvents.add(persistResource(recurring));
     saveCursor(DateTime.parse("2003-10-02T00:00:00Z"));
-    runMapreduce();
+    runAction();
     List<DomainHistory> persistedEntries =
         getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertThat(persistedEntries).hasSize(2);
@@ -502,139 +517,153 @@ public class ExpandRecurringBillingEventsActionTest
     for (int year = 0; year < 2; year++) {
       assertHistoryEntryMatches(
           domain, persistedEntries.get(year), "TheRegistrar", billingDate.plusYears(year), true);
-      expectedEvents.add(defaultOneTimeBuilder()
+      expectedEvents.add(
+          defaultOneTimeBuilder()
               .setBillingTime(billingDate.plusYears(year))
               .setParent(persistedEntries.get(year))
               .setEventTime(eventDate.plusYears(year))
-          .setSyntheticCreationTime(testTime)
               .build());
     }
     assertBillingEventsForResource(domain, Iterables.toArray(expectedEvents, BillingEvent.class));
-    assertCursorAt(testTime);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_singleEvent_beforeRenewal() throws Exception {
-    DateTime testTime = DateTime.parse("2000-01-04T00:00:00Z");
+    // Need to restore to the time before the clock was advanced so that the commit log's timestamp
-    clock.setTo(testTime);
+    // is not inverted when the clock is later reverted.
+    clock.setTo(currentTestTime);
     persistResource(recurring);
+    clock.setTo(DateTime.parse("2000-01-04T00:00:00Z"));
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     // No new history entries should be generated
     assertThat(getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW)).isEmpty();
     assertBillingEventsForResource(domain, recurring);
-    assertCursorAt(testTime);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_singleEvent_afterRecurrenceEnd_inAutorenewGracePeriod() throws Exception {
     // The domain creation date is 1999-01-05, and the first renewal date is thus 2000-01-05.
-    DateTime testTime = DateTime.parse("2001-02-06T00:00:00Z");
+    clock.setTo(DateTime.parse("2001-02-06T00:00:00Z"));
-    clock.setTo(testTime);
+    recurring =
-    recurring = persistResource(recurring.asBuilder()
+        persistResource(
-        // The domain deletion date is 2000-01-29, which is within the 45 day autorenew grace period
+            recurring
+                .asBuilder()
+                // The domain deletion date is 2000-01-29, which is within the 45 day autorenew
+                // grace period
                 // from the renewal date.
                 .setRecurrenceEndTime(DateTime.parse("2000-01-29T00:00:00Z"))
                 .setEventTime(domain.getCreationTime().plusYears(1))
                 .build());
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     DomainHistory persistedEntry =
         getOnlyHistoryEntryOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertHistoryEntryMatches(
         domain, persistedEntry, "TheRegistrar", DateTime.parse("2000-02-19T00:00:00Z"), false);
-    BillingEvent.OneTime expected = defaultOneTimeBuilder()
+    BillingEvent.OneTime expected =
+        defaultOneTimeBuilder()
             .setBillingTime(DateTime.parse("2000-02-19T00:00:00Z"))
             .setParent(persistedEntry)
-        .setSyntheticCreationTime(testTime)
             .build();
     assertBillingEventsForResource(domain, recurring, expected);
-    assertCursorAt(testTime);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_singleEvent_afterRecurrenceEnd_outsideAutorenewGracePeriod() throws Exception {
     // The domain creation date is 1999-01-05, and the first renewal date is thus 2000-01-05.
-    DateTime testTime = DateTime.parse("2001-02-06T00:00:00Z");
+    clock.setTo(DateTime.parse("2001-02-06T00:00:00Z"));
-    clock.setTo(testTime);
+    recurring =
-    recurring = persistResource(recurring.asBuilder()
+        persistResource(
-        // The domain deletion date is 2000-04-05, which is not within the 45 day autorenew grace
+            recurring
+                .asBuilder()
+                // The domain deletion date is 2000-04-05, which is not within the 45 day autorenew
+                // grace
                 // period from the renewal date.
                 .setRecurrenceEndTime(DateTime.parse("2000-04-05T00:00:00Z"))
                 .setEventTime(domain.getCreationTime().plusYears(1))
                 .build());
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     DomainHistory persistedEntry =
         getOnlyHistoryEntryOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertHistoryEntryMatches(
         domain, persistedEntry, "TheRegistrar", DateTime.parse("2000-02-19T00:00:00Z"), true);
-    BillingEvent.OneTime expected = defaultOneTimeBuilder()
+    BillingEvent.OneTime expected =
+        defaultOneTimeBuilder()
             .setBillingTime(DateTime.parse("2000-02-19T00:00:00Z"))
             .setParent(persistedEntry)
-        .setSyntheticCreationTime(testTime)
             .build();
     assertBillingEventsForResource(domain, recurring, expected);
-    assertCursorAt(testTime);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_billingTimeOnLeapYear() throws Exception {
     recurring =
         persistResource(
             recurring.asBuilder().setEventTime(DateTime.parse("2000-01-15T00:00:00Z")).build());
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     DomainHistory persistedEntry =
         getOnlyHistoryEntryOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertHistoryEntryMatches(
         domain, persistedEntry, "TheRegistrar", DateTime.parse("2000-02-29T00:00:00Z"), true);
-    BillingEvent.OneTime expected = defaultOneTimeBuilder()
+    BillingEvent.OneTime expected =
+        defaultOneTimeBuilder()
             .setBillingTime(DateTime.parse("2000-02-29T00:00:00Z"))
             .setEventTime(DateTime.parse("2000-01-15T00:00:00Z"))
             .setParent(persistedEntry)
             .build();
     assertBillingEventsForResource(domain, expected, recurring);
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_expandSingleEvent_billingTimeNotOnLeapYear() throws Exception {
-    DateTime testTime = DateTime.parse("2001-12-01T00:00:00Z");
     recurring =
         persistResource(
             recurring.asBuilder().setEventTime(DateTime.parse("2001-01-15T00:00:00Z")).build());
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    clock.setTo(testTime);
+    clock.setTo(DateTime.parse("2001-12-01T00:00:00Z"));
-    runMapreduce();
+    runAction();
     DomainHistory persistedEntry =
         getOnlyHistoryEntryOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertHistoryEntryMatches(
         domain, persistedEntry, "TheRegistrar", DateTime.parse("2001-03-01T00:00:00Z"), true);
-    BillingEvent.OneTime expected = defaultOneTimeBuilder()
+    BillingEvent.OneTime expected =
+        defaultOneTimeBuilder()
             .setBillingTime(DateTime.parse("2001-03-01T00:00:00Z"))
             .setEventTime(DateTime.parse("2001-01-15T00:00:00Z"))
             .setParent(persistedEntry)
-        .setSyntheticCreationTime(testTime)
             .build();
     assertBillingEventsForResource(domain, expected, recurring);
-    assertCursorAt(testTime);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestSqlOnly
   void testSuccess_expandMultipleEvents() throws Exception {
     persistResource(recurring);
-    BillingEvent.Recurring recurring2 = persistResource(recurring.asBuilder()
+    BillingEvent.Recurring recurring2 =
+        persistResource(
+            recurring
+                .asBuilder()
                 .setEventTime(recurring.getEventTime().plusMonths(3))
                 .setId(3L)
                 .build());
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     List<DomainHistory> persistedEntries =
         getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertThat(persistedEntries).hasSize(2);
     assertHistoryEntryMatches(
-        domain, persistedEntries.get(0), "TheRegistrar", DateTime.parse("2000-02-19T00:00:00Z"),
+        domain,
+        persistedEntries.get(0),
+        "TheRegistrar",
+        DateTime.parse("2000-02-19T00:00:00Z"),
         true);
     BillingEvent.OneTime expected =
         defaultOneTimeBuilder()
@@ -642,7 +671,10 @@ public class ExpandRecurringBillingEventsActionTest
             .setCancellationMatchingBillingEvent(recurring.createVKey())
             .build();
     assertHistoryEntryMatches(
-        domain, persistedEntries.get(1), "TheRegistrar", DateTime.parse("2000-05-20T00:00:00Z"),
+        domain,
+        persistedEntries.get(1),
+        "TheRegistrar",
+        DateTime.parse("2000-05-20T00:00:00Z"),
         true);
     BillingEvent.OneTime expected2 =
         defaultOneTimeBuilder()
@@ -652,10 +684,10 @@ public class ExpandRecurringBillingEventsActionTest
             .setCancellationMatchingBillingEvent(recurring2.createVKey())
             .build();
     assertBillingEventsForResource(domain, expected, expected2, recurring, recurring2);
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_premiumDomain() throws Exception {
     persistResource(
         Registry.get("tld")
@@ -664,86 +696,87 @@ public class ExpandRecurringBillingEventsActionTest
             .build());
     persistResource(recurring);
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     DomainHistory persistedEntry =
         getOnlyHistoryEntryOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertHistoryEntryMatches(
         domain, persistedEntry, "TheRegistrar", DateTime.parse("2000-02-19T00:00:00Z"), true);
-    BillingEvent.OneTime expected = defaultOneTimeBuilder()
+    BillingEvent.OneTime expected =
-        .setParent(persistedEntry)
+        defaultOneTimeBuilder().setParent(persistedEntry).setCost(Money.of(USD, 100)).build();
-        .setCost(Money.of(USD, 100))
-        .build();
     assertBillingEventsForResource(domain, expected, recurring);
-    assertCursorAt(beginningOfTest);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testSuccess_varyingRenewPrices() throws Exception {
-    DateTime testTime = beginningOfTest.plusYears(1);
+    clock.setTo(currentTestTime);
     persistResource(
         Registry.get("tld")
             .asBuilder()
             .setRenewBillingCostTransitions(
                 ImmutableSortedMap.of(
-                    START_OF_TIME, Money.of(USD, 8),
+                    START_OF_TIME,
-                    DateTime.parse("2000-06-01T00:00:00Z"), Money.of(USD, 10)))
+                    Money.of(USD, 8),
+                    DateTime.parse("2000-06-01T00:00:00Z"),
+                    Money.of(USD, 10)))
             .build());
-    clock.setTo(testTime);
+    clock.setTo(DateTime.parse("2001-10-02T00:00:00Z"));
     persistResource(recurring);
     action.cursorTimeParam = Optional.of(START_OF_TIME);
-    runMapreduce();
+    runAction();
     List<DomainHistory> persistedEntries =
         getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW, DomainHistory.class);
     assertThat(persistedEntries).hasSize(2);
     DateTime eventDate = DateTime.parse("2000-01-05T00:00:00Z");
     DateTime billingDate = DateTime.parse("2000-02-19T00:00:00Z");
     assertHistoryEntryMatches(domain, persistedEntries.get(0), "TheRegistrar", billingDate, true);
-    BillingEvent.OneTime cheaper = defaultOneTimeBuilder()
+    BillingEvent.OneTime cheaper =
+        defaultOneTimeBuilder()
             .setBillingTime(billingDate)
             .setEventTime(eventDate)
             .setParent(persistedEntries.get(0))
             .setCost(Money.of(USD, 8))
-        .setSyntheticCreationTime(testTime)
             .build();
     assertHistoryEntryMatches(
         domain, persistedEntries.get(1), "TheRegistrar", billingDate.plusYears(1), true);
-    BillingEvent.OneTime expensive = cheaper.asBuilder()
+    BillingEvent.OneTime expensive =
+        cheaper
+            .asBuilder()
             .setCost(Money.of(USD, 10))
             .setBillingTime(billingDate.plusYears(1))
             .setEventTime(eventDate.plusYears(1))
             .setParent(persistedEntries.get(1))
             .build();
     assertBillingEventsForResource(domain, recurring, cheaper, expensive);
-    assertCursorAt(testTime);
+    assertCursorAt(currentTestTime);
   }
 
-  @Test
+  @TestOfyAndSql
   void testFailure_cursorAfterExecutionTime() {
     action.cursorTimeParam = Optional.of(clock.nowUtc().plusYears(1));
-    IllegalArgumentException thrown =
+    IllegalArgumentException thrown = assertThrows(IllegalArgumentException.class, this::runAction);
-        assertThrows(IllegalArgumentException.class, this::runMapreduce);
     assertThat(thrown)
         .hasMessageThat()
         .contains("Cursor time must be earlier than execution time.");
   }
 
-  @Test
+  @TestOfyAndSql
   void testFailure_cursorAtExecutionTime() {
     // The clock advances one milli on runMapreduce.
     action.cursorTimeParam = Optional.of(clock.nowUtc().plusMillis(1));
-    IllegalArgumentException thrown =
+    IllegalArgumentException thrown = assertThrows(IllegalArgumentException.class, this::runAction);
-        assertThrows(IllegalArgumentException.class, this::runMapreduce);
     assertThat(thrown)
         .hasMessageThat()
         .contains("Cursor time must be earlier than execution time.");
   }
 
-  @Test
+  @TestOfyOnly
   void testFailure_mapperException_doesNotMoveCursor() throws Exception {
     saveCursor(START_OF_TIME); // Need a saved cursor to verify that it didn't move.
+    clock.advanceOneMilli();
     // Set target to a TLD that doesn't exist.
     recurring = persistResource(recurring.asBuilder().setTargetId("domain.junk").build());
-    runMapreduce();
+    runAction();
     // No new history entries should be generated
     assertThat(getHistoryEntriesOfType(domain, DOMAIN_AUTORENEW)).isEmpty();
     assertBillingEvents(recurring); // only the bogus one in Datastore

db/src/main/resources/sql/flyway.txt

@@ -94,3 +94,4 @@ V93__defer_all_fkeys.sql
 V94__rename_lock_scope.sql
 V95__add_contacts_indexes_on_domain.sql
 V96__rename_sql_checkpoint_fields.sql
+V97__add_recurrence_history_id_column_to_onetime.sql

db/src/main/resources/sql/flyway/V97__add_recurrence_history_id_column_to_onetime.sql

@@ -0,0 +1,21 @@
+-- Copyright 2021 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.
+
+ALTER TABLE "BillingEvent"
+    ADD COLUMN IF NOT EXISTS "recurrence_history_revision_id" INT8;
+
+ALTER TABLE "BillingEvent" ADD CONSTRAINT fk_billing_event_recurrence_history
+   FOREIGN KEY (domain_repo_id, recurrence_history_revision_id)
+   REFERENCES "DomainHistory"(domain_repo_id, history_revision_id)
+   DEFERRABLE INITIALLY DEFERRED;

db/src/main/resources/sql/schema/db-schema.sql.generated

@@ -63,6 +63,7 @@
         cost_amount numeric(19, 2),
         cost_currency text,
         period_years int4,
+        recurrence_history_revision_id int8,
         synthetic_creation_time timestamptz,
         primary key (billing_event_id)
     );

db/src/main/resources/sql/schema/nomulus.golden.sql

@@ -97,7 +97,8 @@ CREATE TABLE public."BillingEvent" (
     cost_amount numeric(19,2),
     cost_currency text,
     period_years integer,
-    synthetic_creation_time timestamp with time zone
+    synthetic_creation_time timestamp with time zone,
+    recurrence_history_revision_id bigint
 );
 
 
@@ -1985,6 +1986,14 @@ ALTER TABLE ONLY public."BillingEvent"
     ADD CONSTRAINT fk_billing_event_domain_history FOREIGN KEY (domain_repo_id, domain_history_revision_id) REFERENCES public."DomainHistory"(domain_repo_id, history_revision_id) DEFERRABLE INITIALLY DEFERRED;
 
 
+--
+-- Name: BillingEvent fk_billing_event_recurrence_history; Type: FK CONSTRAINT; Schema: public; Owner: -
+--
+
+ALTER TABLE ONLY public."BillingEvent"
+    ADD CONSTRAINT fk_billing_event_recurrence_history FOREIGN KEY (domain_repo_id, recurrence_history_revision_id) REFERENCES public."DomainHistory"(domain_repo_id, history_revision_id) DEFERRABLE INITIALLY DEFERRED;
+
+
 --
 -- Name: BillingEvent fk_billing_event_registrar_id; Type: FK CONSTRAINT; Schema: public; Owner: -
 --