Refactor Fee handling so that each fee knows if it's premium (#626)

* Refactor Fee handling so that each fee knows if it's premium

This is a noop for now, as the new isPremium boolean isn't yet used by anything,
but it will be used in follow-up PRs to add additional fee information using the
fee extension (see: b/157621273).  Specifically what we're trying to do here is
return <fee:command name="create" standard="1"> (using the finalized version of
the fee extension) when an entire command has no premium fee associated with
it. And in the current earlier versions of the fee extension that we support,
we'll want to display the correct fee amount and class for creates/checks on
reserved domains when a valid allocation token is passed. This also needs the
isPremium information.

There are no testing implications yet because isPremium isn't exposed anywhere,
but there will definitely be lots of test changes once it's feeding into EPP-
visible changes.

* Rename things, add method Javadoc

* Apply formatting

core/src/main/java/google/registry/backup/AppEngineEnvironment.java

@@ -0,0 +1,66 @@
+// Copyright 2020 The Nomulus Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package google.registry.backup;
+
+import com.google.apphosting.api.ApiProxy;
+import com.google.apphosting.api.ApiProxy.Environment;
+import com.google.common.collect.ImmutableMap;
+import java.io.Closeable;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+
+/**
+ * Sets up a placeholder {@link Environment} on a non-AppEngine platform so that Datastore Entities
+ * can be deserialized. See {@code DatastoreEntityExtension} in test source for more information.
+ */
+public class AppEngineEnvironment implements Closeable {
+
+  private static final Environment PLACEHOLDER_ENV = createAppEngineEnvironment();
+
+  private boolean isPlaceHolderNeeded;
+
+  AppEngineEnvironment() {
+    isPlaceHolderNeeded = ApiProxy.getCurrentEnvironment() == null;
+    // isPlaceHolderNeeded may be true when we are invoked in a test with AppEngineRule.
+    if (isPlaceHolderNeeded) {
+      ApiProxy.setEnvironmentForCurrentThread(PLACEHOLDER_ENV);
+    }
+  }
+
+  @Override
+  public void close() {
+    if (isPlaceHolderNeeded) {
+      ApiProxy.setEnvironmentForCurrentThread(null);
+    }
+  }
+
+  /** Returns a placeholder {@link Environment} that can return hardcoded AppId and Attributes. */
+  private static Environment createAppEngineEnvironment() {
+    return (Environment)
+        Proxy.newProxyInstance(
+            Environment.class.getClassLoader(),
+            new Class[] {Environment.class},
+            (Object proxy, Method method, Object[] args) -> {
+              switch (method.getName()) {
+                case "getAppId":
+                  return "PlaceholderAppId";
+                case "getAttributes":
+                  return ImmutableMap.<String, Object>of();
+                default:
+                  throw new UnsupportedOperationException(method.getName());
+              }
+            });
+  }
+}

core/src/main/java/google/registry/backup/CommitLogImports.java

@@ -0,0 +1,95 @@
+// Copyright 2020 The Nomulus Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package google.registry.backup;
+
+import static com.google.common.base.Preconditions.checkState;
+import static google.registry.backup.BackupUtils.createDeserializingIterator;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Streams;
+import google.registry.model.ImmutableObject;
+import google.registry.model.ofy.CommitLogCheckpoint;
+import google.registry.model.ofy.CommitLogManifest;
+import google.registry.model.ofy.CommitLogMutation;
+import java.io.BufferedInputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.util.Iterator;
+import java.util.stream.Stream;
+
+/**
+ * Helpers for reading CommitLog records from a file.
+ *
+ * <p>This class is adapted from {@link RestoreCommitLogsAction}, and will be used in the initial
+ * population of the Cloud SQL database.
+ */
+public final class CommitLogImports {
+
+  private CommitLogImports() {}
+
+  /**
+   * Returns entities in an {@code inputStream} (from a single CommitLog file) as an {@link
+   * ImmutableList} of {@link VersionedEntity} records. Upon completion the {@code inputStream} is
+   * closed.
+   *
+   * <p>The returned list may be empty, since CommitLogs are written at fixed intervals regardless
+   * if actual changes exist.
+   *
+   * <p>A CommitLog file starts with a {@link CommitLogCheckpoint}, followed by (repeated)
+   * subsequences of [{@link CommitLogManifest}, [{@link CommitLogMutation}] ...]. Each subsequence
+   * represents the changes in one transaction. The {@code CommitLogManifest} contains deleted
+   * entity keys, whereas each {@code CommitLogMutation} contains one whole entity.
+   */
+  public static ImmutableList<VersionedEntity> loadEntities(InputStream inputStream) {
+    try (AppEngineEnvironment appEngineEnvironment = new AppEngineEnvironment();
+        InputStream input = new BufferedInputStream(inputStream)) {
+      Iterator<ImmutableObject> commitLogs = createDeserializingIterator(input);
+      checkState(commitLogs.hasNext());
+      checkState(commitLogs.next() instanceof CommitLogCheckpoint);
+
+      return Streams.stream(commitLogs)
+          .map(
+              e ->
+                  e instanceof CommitLogManifest
+                      ? VersionedEntity.fromManifest((CommitLogManifest) e)
+                      : Stream.of(VersionedEntity.fromMutation((CommitLogMutation) e)))
+          .flatMap(s -> s)
+          .collect(ImmutableList.toImmutableList());
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /** Covenience method that adapts {@link #loadEntities(InputStream)} to a {@link File}. */
+  public static ImmutableList<VersionedEntity> loadEntities(File commitLogFile) {
+    try {
+      return loadEntities(new FileInputStream(commitLogFile));
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Covenience method that adapts {@link #loadEntities(InputStream)} to a {@link
+   * ReadableByteChannel}.
+   */
+  public static ImmutableList<VersionedEntity> loadEntities(ReadableByteChannel channel) {
+    return loadEntities(Channels.newInputStream(channel));
+  }
+}

core/src/main/java/google/registry/backup/VersionedEntity.java

@@ -0,0 +1,160 @@
+// Copyright 2020 The Nomulus Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package google.registry.backup;
+
+import com.google.appengine.api.datastore.Entity;
+import com.google.appengine.api.datastore.EntityTranslator;
+import com.google.appengine.api.datastore.Key;
+import com.google.auto.value.AutoValue;
+import com.google.auto.value.extension.memoized.Memoized;
+import google.registry.model.ofy.CommitLogManifest;
+import google.registry.model.ofy.CommitLogMutation;
+import java.io.Serializable;
+import java.util.Arrays;
+import java.util.Optional;
+import java.util.stream.Stream;
+import javax.annotation.Nullable;
+
+/**
+ * A Datastore {@link Entity Entity's} timestamped state.
+ *
+ * <p>For a new or updated Entity, its ProtocolBuffer bytes are stored along with its {@link Key}.
+ * For a deleted entity, only its {@link Key} is stored, and the {@link #entityProtoBytes} is left
+ * as null.
+ *
+ * <p>Note that {@link Optional java.util.Optional} is not serializable, therefore cannot be used as
+ * property type in this class.
+ */
+@AutoValue
+public abstract class VersionedEntity implements Serializable {
+
+  private static final long serialVersionUID = 1L;
+
+  public abstract long commitTimeMills();
+
+  /** The {@link Key} of the {@link Entity}. */
+  public abstract Key key();
+
+  /** Serialized form of the {@link Entity}. This property is {@code null} for a deleted Entity. */
+  @Nullable
+  abstract ImmutableBytes entityProtoBytes();
+
+  @Memoized
+  public Optional<Entity> getEntity() {
+    return Optional.ofNullable(entityProtoBytes())
+        .map(ImmutableBytes::getBytes)
+        .map(EntityTranslator::createFromPbBytes);
+  }
+
+  public boolean isDelete() {
+    return entityProtoBytes() == null;
+  }
+
+  /**
+   * Converts deleted entity keys in {@code manifest} into a {@link Stream} of {@link
+   * VersionedEntity VersionedEntities}. See {@link CommitLogImports#loadEntities} for more
+   * information.
+   */
+  public static Stream<VersionedEntity> fromManifest(CommitLogManifest manifest) {
+    long commitTimeMillis = manifest.getCommitTime().getMillis();
+    return manifest.getDeletions().stream()
+        .map(com.googlecode.objectify.Key::getRaw)
+        .map(key -> builder().commitTimeMills(commitTimeMillis).key(key).build());
+  }
+
+  /* Converts a {@link CommitLogMutation} to a {@link VersionedEntity}. */
+  public static VersionedEntity fromMutation(CommitLogMutation mutation) {
+    return from(
+        com.googlecode.objectify.Key.create(mutation).getParent().getId(),
+        mutation.getEntityProtoBytes());
+  }
+
+  public static VersionedEntity from(long commitTimeMillis, byte[] entityProtoBytes) {
+    return builder()
+        .entityProtoBytes(entityProtoBytes)
+        .key(EntityTranslator.createFromPbBytes(entityProtoBytes).getKey())
+        .commitTimeMills(commitTimeMillis)
+        .build();
+  }
+
+  static Builder builder() {
+    return new AutoValue_VersionedEntity.Builder();
+  }
+
+  @AutoValue.Builder
+  public abstract static class Builder {
+
+    public abstract Builder commitTimeMills(long commitTimeMillis);
+
+    abstract Builder entityProtoBytes(ImmutableBytes bytes);
+
+    public abstract Builder key(Key key);
+
+    public abstract VersionedEntity build();
+
+    public Builder entityProtoBytes(byte[] bytes) {
+      return entityProtoBytes(new ImmutableBytes(bytes));
+    }
+  }
+
+  /**
+   * Wraps a byte array and prevents it from being modified by its original owner.
+   *
+   * <p>While this class seems an overkill, it exists for two reasons:
+   *
+   * <ul>
+   *   <li>It is easier to override the {@link #equals} method here (for value-equivalence check)
+   *       than to override the AutoValue-generated {@code equals} method.
+   *   <li>To appease the style checker, which forbids arrays as AutoValue property.
+   * </ul>
+   */
+  static final class ImmutableBytes implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private final byte[] bytes;
+
+    ImmutableBytes(byte[] bytes) {
+      this.bytes = Arrays.copyOf(bytes, bytes.length);
+    }
+
+    /**
+     * Returns the saved byte array. Invocation is restricted to trusted callers, who must not
+     * modify the array.
+     */
+    byte[] getBytes() {
+      return bytes;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (!(o instanceof ImmutableBytes)) {
+        return false;
+      }
+      ImmutableBytes that = (ImmutableBytes) o;
+      // Do not use Objects.equals, which checks reference identity instead of data in array.
+      return Arrays.equals(bytes, that.bytes);
+    }
+
+    @Override
+    public int hashCode() {
+      // Do not use Objects.hashCode, which hashes the reference, not the data in array.
+      return Arrays.hashCode(bytes);
+    }
+  }
+}

core/src/main/java/google/registry/beam/initsql/BackupPaths.java

@@ -17,8 +17,7 @@ package google.registry.beam.initsql;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Strings.isNullOrEmpty;
 
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
+import org.joda.time.DateTime;
 
 /**
  * Helpers for determining the fully qualified paths to Nomulus backup files. A backup consists of a
@@ -30,13 +29,9 @@ public final class BackupPaths {
 
   private static final String WILDCARD_CHAR = "*";
   private static final String EXPORT_PATTERN_TEMPLATE = "%s/all_namespaces/kind_%s/input-%s";
-  /**
-   * Regex pattern that captures the kind string in a file name. Datastore places no restrictions on
-   * what characters may be used in a kind string.
-   */
-  private static final String FILENAME_TO_KIND_REGEX = ".+/all_namespaces/kind_(.+)/input-.+";
 
-  private static final Pattern FILENAME_TO_KIND_PATTERN = Pattern.compile(FILENAME_TO_KIND_REGEX);
+  public static final String COMMIT_LOG_NAME_PREFIX = "commit_diff_until_";
+  private static final String COMMIT_LOG_PATTERN_TEMPLATE = "%s/" + COMMIT_LOG_NAME_PREFIX + "*";
 
   /**
    * Returns a regex pattern that matches all Datastore export files of a given {@code kind}.
@@ -65,22 +60,15 @@ public final class BackupPaths {
     return String.format(EXPORT_PATTERN_TEMPLATE, exportDir, kind, Integer.toString(shard));
   }
 
-  /**
-   * Returns the 'kind' of entity stored in a file based on the file name.
-   *
-   * <p>This method poses low risk and greatly simplifies the implementation of some transforms in
-   * {@link ExportLoadingTransforms}.
-   *
-   * @see ExportLoadingTransforms
-   */
-  public static String getKindFromFileName(String fileName) {
+  public static String getCommitLogFileNamePattern(String commitLogDir) {
+    return String.format(COMMIT_LOG_PATTERN_TEMPLATE, commitLogDir);
+  }
+
+  /** Gets the Commit timestamp from a CommitLog file name. */
+  public static DateTime getCommitLogTimestamp(String fileName) {
     checkArgument(!isNullOrEmpty(fileName), "Null or empty fileName.");
-    Matcher matcher = FILENAME_TO_KIND_PATTERN.matcher(fileName);
-    checkArgument(
-        matcher.matches(),
-        "Illegal file name %s, should match %s.",
-        fileName,
-        FILENAME_TO_KIND_REGEX);
-    return matcher.group(1);
+    int start = fileName.lastIndexOf(COMMIT_LOG_NAME_PREFIX);
+    checkArgument(start >= 0, "Illegal file name %s.", fileName);
+    return DateTime.parse(fileName.substring(start + COMMIT_LOG_NAME_PREFIX.length()));
   }
 }

core/src/main/java/google/registry/beam/initsql/CommitLogTransforms.java

@@ -0,0 +1,109 @@
+// Copyright 2020 The Nomulus Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package google.registry.beam.initsql;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static google.registry.beam.initsql.BackupPaths.getCommitLogFileNamePattern;
+import static google.registry.beam.initsql.BackupPaths.getCommitLogTimestamp;
+import static google.registry.beam.initsql.Transforms.processFiles;
+import static google.registry.util.DateTimeUtils.isBeforeOrAt;
+
+import google.registry.backup.CommitLogImports;
+import google.registry.backup.VersionedEntity;
+import java.io.IOException;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.FileIO.ReadableFile;
+import org.apache.beam.sdk.io.fs.MatchResult.Metadata;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.DateTime;
+
+/**
+ * {@link org.apache.beam.sdk.transforms.PTransform Pipeline transforms} for loading from Nomulus
+ * CommitLog files. They are all part of a transformation that loads raw records from a sequence of
+ * Datastore CommitLog files, and are broken apart for testing.
+ */
+public class CommitLogTransforms {
+
+  /**
+   * Returns a {@link PTransform transform} that can generate a collection of patterns that match
+   * all Datastore CommitLog files.
+   */
+  public static PTransform<PBegin, PCollection<String>> getCommitLogFilePatterns(
+      String commitLogDir) {
+    return Create.of(getCommitLogFileNamePattern(commitLogDir)).withCoder(StringUtf8Coder.of());
+  }
+
+  /**
+   * Returns files with timestamps between {@code fromTime} (inclusive) and {@code endTime}
+   * (exclusive).
+   */
+  public static PTransform<PCollection<? extends String>, PCollection<String>>
+      filterCommitLogsByTime(DateTime fromTime, DateTime toTime) {
+    checkNotNull(fromTime, "fromTime");
+    checkNotNull(toTime, "toTime");
+    checkArgument(
+        fromTime.isBefore(toTime),
+        "Invalid time range: fromTime (%s) is before endTime (%s)",
+        fromTime,
+        toTime);
+    return ParDo.of(new FilterCommitLogFileByTime(fromTime, toTime));
+  }
+
+  /** Returns a {@link PTransform} from file {@link Metadata} to {@link VersionedEntity}. */
+  public static PTransform<PCollection<Metadata>, PCollection<VersionedEntity>>
+      loadCommitLogsFromFiles() {
+    return processFiles(new LoadOneCommitLogsFile());
+  }
+
+  static class FilterCommitLogFileByTime extends DoFn<String, String> {
+    private final DateTime fromTime;
+    private final DateTime toTime;
+
+    public FilterCommitLogFileByTime(DateTime fromTime, DateTime toTime) {
+      this.fromTime = fromTime;
+      this.toTime = toTime;
+    }
+
+    @ProcessElement
+    public void processElement(@Element String fileName, OutputReceiver<String> out) {
+      DateTime timestamp = getCommitLogTimestamp(fileName);
+      if (isBeforeOrAt(fromTime, timestamp) && timestamp.isBefore(toTime)) {
+        out.output(fileName);
+      }
+    }
+  }
+
+  /**
+   * Reads a CommitLog file and converts its content into {@link VersionedEntity VersionedEntities}.
+   */
+  static class LoadOneCommitLogsFile extends DoFn<ReadableFile, VersionedEntity> {
+
+    @ProcessElement
+    public void processElement(@Element ReadableFile file, OutputReceiver<VersionedEntity> out) {
+      try {
+        CommitLogImports.loadEntities(file.open()).forEach(out::output);
+      } catch (IOException e) {
+        // Let the pipeline retry the whole file.
+        throw new RuntimeException(e);
+      }
+    }
+  }
+}

core/src/main/java/google/registry/beam/initsql/ExportLoadingTransforms.java

@@ -15,38 +15,26 @@
 package google.registry.beam.initsql;
 
 import static google.registry.beam.initsql.BackupPaths.getExportFileNamePattern;
-import static google.registry.beam.initsql.BackupPaths.getKindFromFileName;
-import static org.apache.beam.sdk.values.TypeDescriptors.kvs;
-import static org.apache.beam.sdk.values.TypeDescriptors.strings;
+import static google.registry.beam.initsql.Transforms.processFiles;
 
 import com.google.common.collect.ImmutableList;
+import google.registry.backup.VersionedEntity;
 import google.registry.tools.LevelDbLogReader;
 import java.io.IOException;
 import java.util.Collection;
 import org.apache.beam.sdk.coders.StringUtf8Coder;
-import org.apache.beam.sdk.io.Compression;
-import org.apache.beam.sdk.io.FileIO;
 import org.apache.beam.sdk.io.FileIO.ReadableFile;
-import org.apache.beam.sdk.io.fs.EmptyMatchTreatment;
 import org.apache.beam.sdk.io.fs.MatchResult.Metadata;
 import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.transforms.DoFn;
-import org.apache.beam.sdk.transforms.MapElements;
 import org.apache.beam.sdk.transforms.PTransform;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PBegin;
 import org.apache.beam.sdk.values.PCollection;
-import org.apache.beam.sdk.values.TypeDescriptor;
 
 /**
  * {@link PTransform Pipeline transforms} for loading from Datastore export files. They are all part
  * of a transformation that loads raw records from a Datastore export, and are broken apart for
  * testing.
- *
- * <p>We drop the 'kind' information in {@link #getDatastoreExportFilePatterns} and recover it later
- * using the file paths. Although we could have kept it by passing around {@link KV key-value
- * pairs}, the code would be more complicated, especially in {@link #loadDataFromFiles()}.
  */
 public class ExportLoadingTransforms {
 
@@ -63,51 +51,28 @@ public class ExportLoadingTransforms {
         .withCoder(StringUtf8Coder.of());
   }
 
-  /**
-   * Returns a {@link PTransform} from file name patterns to file {@link Metadata Metadata records}.
-   */
-  public static PTransform<PCollection<String>, PCollection<Metadata>> getFilesByPatterns() {
-    return new PTransform<PCollection<String>, PCollection<Metadata>>() {
-      @Override
-      public PCollection<Metadata> expand(PCollection<String> input) {
-        return input.apply(FileIO.matchAll().withEmptyMatchTreatment(EmptyMatchTreatment.DISALLOW));
-      }
-    };
+  /** Returns a {@link PTransform} from file {@link Metadata} to {@link VersionedEntity}. */
+  public static PTransform<PCollection<Metadata>, PCollection<VersionedEntity>>
+      loadExportDataFromFiles() {
+    return processFiles(new LoadOneExportShard());
   }
 
   /**
-   * Returns a {@link PTransform} from file {@link Metadata} to {@link KV Key-Value pairs} with
-   * entity 'kind' as key and raw record as value.
-   */
-  public static PTransform<PCollection<Metadata>, PCollection<KV<String, byte[]>>>
-      loadDataFromFiles() {
-    return new PTransform<PCollection<Metadata>, PCollection<KV<String, byte[]>>>() {
-      @Override
-      public PCollection<KV<String, byte[]>> expand(PCollection<Metadata> input) {
-        return input
-            .apply(FileIO.readMatches().withCompression(Compression.UNCOMPRESSED))
-            .apply(
-                MapElements.into(kvs(strings(), TypeDescriptor.of(ReadableFile.class)))
-                    .via(file -> KV.of(getKindFromFileName(file.getMetadata().toString()), file)))
-            .apply("Load One LevelDb File", ParDo.of(new LoadOneFile()));
-      }
-    };
-  }
-
-  /**
-   * Reads a LevelDb file and converts each raw record into a {@link KV pair} of kind and bytes.
+   * Reads a LevelDb file and converts each raw record into a {@link VersionedEntity}. All such
+   * entities use {@link Long#MIN_VALUE} as timestamp, so that they go before data from CommitLogs.
    *
    * <p>LevelDb files are not seekable because a large object may span multiple blocks. If a
    * sequential read fails, the file needs to be retried from the beginning.
    */
-  private static class LoadOneFile extends DoFn<KV<String, ReadableFile>, KV<String, byte[]>> {
+  private static class LoadOneExportShard extends DoFn<ReadableFile, VersionedEntity> {
+
+    private static final long TIMESTAMP = Long.MIN_VALUE;
 
     @ProcessElement
-    public void processElement(
-        @Element KV<String, ReadableFile> kv, OutputReceiver<KV<String, byte[]>> output) {
+    public void processElement(@Element ReadableFile file, OutputReceiver<VersionedEntity> output) {
       try {
-        LevelDbLogReader.from(kv.getValue().open())
-            .forEachRemaining(record -> output.output(KV.of(kv.getKey(), record)));
+        LevelDbLogReader.from(file.open())
+            .forEachRemaining(record -> output.output(VersionedEntity.from(TIMESTAMP, record)));
       } catch (IOException e) {
         // Let the pipeline retry the whole file.
         throw new RuntimeException(e);

core/src/main/java/google/registry/beam/initsql/Transforms.java

@@ -0,0 +1,61 @@
+// Copyright 2020 The Nomulus Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package google.registry.beam.initsql;
+
+import google.registry.backup.VersionedEntity;
+import org.apache.beam.sdk.io.Compression;
+import org.apache.beam.sdk.io.FileIO;
+import org.apache.beam.sdk.io.FileIO.ReadableFile;
+import org.apache.beam.sdk.io.fs.EmptyMatchTreatment;
+import org.apache.beam.sdk.io.fs.MatchResult.Metadata;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * Common {@link PTransform pipeline transforms} used in pipelines that load from both Datastore
+ * export files and Nomulus CommitLog files.
+ */
+public class Transforms {
+
+  /**
+   * Returns a {@link PTransform} from file name patterns to file {@link Metadata Metadata records}.
+   */
+  public static PTransform<PCollection<String>, PCollection<Metadata>> getFilesByPatterns() {
+    return new PTransform<PCollection<String>, PCollection<Metadata>>() {
+      @Override
+      public PCollection<Metadata> expand(PCollection<String> input) {
+        return input.apply(FileIO.matchAll().withEmptyMatchTreatment(EmptyMatchTreatment.DISALLOW));
+      }
+    };
+  }
+
+  /**
+   * Returns a {@link PTransform} from file {@link Metadata} to {@link VersionedEntity} using
+   * caller-provided {@code transformer}.
+   */
+  public static PTransform<PCollection<Metadata>, PCollection<VersionedEntity>> processFiles(
+      DoFn<ReadableFile, VersionedEntity> transformer) {
+    return new PTransform<PCollection<Metadata>, PCollection<VersionedEntity>>() {
+      @Override
+      public PCollection<VersionedEntity> expand(PCollection<Metadata> input) {
+        return input
+            .apply(FileIO.readMatches().withCompression(Compression.UNCOMPRESSED))
+            .apply(transformer.getClass().getSimpleName(), ParDo.of(transformer));
+      }
+    };
+  }
+}

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

@@ -690,7 +690,7 @@ public class DomainFlowUtils {
     List<Fee> fees = feeCommand.get().getFees();
     // The schema guarantees that at least one fee will be present.
     checkState(!fees.isEmpty());
-    BigDecimal total = BigDecimal.ZERO;
+    BigDecimal total = zeroInCurrency(feeCommand.get().getCurrency());
     for (Fee fee : fees) {
       if (!fee.hasDefaultAttributes()) {
         throw new UnsupportedFeeAttributeException();
@@ -938,6 +938,16 @@ public class DomainFlowUtils {
     }
   }
 
+  /**
+   * Returns zero for a specific currency.
+   *
+   * <p>{@link BigDecimal} has a concept of significant figures, so zero is not always zero. E.g.
+   * zero in USD is 0.00, whereas zero in Yen is 0, and zero in Dinars is 0.000 (!).
+   */
+  static BigDecimal zeroInCurrency(CurrencyUnit currencyUnit) {
+    return Money.of(currencyUnit, BigDecimal.ZERO).getAmount();
+  }
+
   /**
    * Check that if there's a claims notice it's on the claims list, and that if there's not one it's
    * not on the claims list.

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

@@ -14,8 +14,9 @@
 
 package google.registry.flows.domain;
 
+import static google.registry.flows.domain.DomainFlowUtils.zeroInCurrency;
 import static google.registry.pricing.PricingEngineProxy.getDomainFeeClass;
-import static google.registry.pricing.PricingEngineProxy.getDomainRenewCost;
+import static google.registry.pricing.PricingEngineProxy.getPricesForDomainName;
 
 import com.google.common.net.InternetDomainName;
 import google.registry.flows.EppException;
@@ -33,7 +34,6 @@ import google.registry.model.domain.fee.Fee;
 import google.registry.model.domain.token.AllocationToken;
 import google.registry.model.pricing.PremiumPricingEngine.DomainPrices;
 import google.registry.model.registry.Registry;
-import google.registry.pricing.PricingEngineProxy;
 import java.math.BigDecimal;
 import java.math.RoundingMode;
 import java.util.Optional;
@@ -64,21 +64,27 @@ public final class DomainPricingLogic {
   public FeesAndCredits getCreatePrice(
       Registry registry,
       String domainName,
-      DateTime date,
+      DateTime dateTime,
       int years,
       boolean isAnchorTenant,
       Optional<AllocationToken> allocationToken)
       throws EppException {
     CurrencyUnit currency = registry.getCurrency();
+
+    BaseFee createFeeOrCredit;
     // Domain create cost is always zero for anchor tenants
-    Money domainCreateCost =
-        isAnchorTenant
-            ? Money.of(currency, BigDecimal.ZERO)
-            : getDomainCreateCostWithDiscount(domainName, date, years, allocationToken);
-    BaseFee createFeeOrCredit = Fee.create(domainCreateCost.getAmount(), FeeType.CREATE);
+    if (isAnchorTenant) {
+      createFeeOrCredit = Fee.create(zeroInCurrency(currency), FeeType.CREATE, false);
+    } else {
+      DomainPrices domainPrices = getPricesForDomainName(domainName, dateTime);
+      Money domainCreateCost =
+          getDomainCreateCostWithDiscount(domainPrices, years, allocationToken);
+      createFeeOrCredit =
+          Fee.create(domainCreateCost.getAmount(), FeeType.CREATE, domainPrices.isPremium());
+    }
 
     // Create fees for the cost and the EAP fee, if any.
-    Fee eapFee = registry.getEapFeeFor(date);
+    Fee eapFee = registry.getEapFeeFor(dateTime);
     FeesAndCredits.Builder feesBuilder =
         new FeesAndCredits.Builder().setCurrency(currency).addFeeOrCredit(createFeeOrCredit);
     // Don't charge anchor tenants EAP fees.
@@ -92,7 +98,7 @@ public final class DomainPricingLogic {
             .setFeesAndCredits(feesBuilder.build())
             .setRegistry(registry)
             .setDomainName(InternetDomainName.from(domainName))
-            .setAsOfDate(date)
+            .setAsOfDate(dateTime)
             .setYears(years)
             .build());
   }
@@ -100,69 +106,73 @@ public final class DomainPricingLogic {
   /** Returns a new renew price for the pricer. */
   @SuppressWarnings("unused")
   public FeesAndCredits getRenewPrice(
-      Registry registry,
-      String domainName,
-      DateTime date,
-      int years)
-      throws EppException {
-    Money renewCost = getDomainRenewCost(domainName, date, years);
+      Registry registry, String domainName, DateTime dateTime, int years) throws EppException {
+    DomainPrices domainPrices = getPricesForDomainName(domainName, dateTime);
+    BigDecimal renewCost = domainPrices.getRenewCost().multipliedBy(years).getAmount();
     return customLogic.customizeRenewPrice(
         RenewPriceParameters.newBuilder()
             .setFeesAndCredits(
                 new FeesAndCredits.Builder()
                     .setCurrency(registry.getCurrency())
-                    .addFeeOrCredit(Fee.create(renewCost.getAmount(), FeeType.RENEW))
+                    .addFeeOrCredit(Fee.create(renewCost, FeeType.RENEW, domainPrices.isPremium()))
                     .build())
             .setRegistry(registry)
             .setDomainName(InternetDomainName.from(domainName))
-            .setAsOfDate(date)
+            .setAsOfDate(dateTime)
             .setYears(years)
             .build());
   }
 
   /** Returns a new restore price for the pricer. */
-  public FeesAndCredits getRestorePrice(Registry registry, String domainName, DateTime date)
+  public FeesAndCredits getRestorePrice(Registry registry, String domainName, DateTime dateTime)
       throws EppException {
+    DomainPrices domainPrices = getPricesForDomainName(domainName, dateTime);
     FeesAndCredits feesAndCredits =
         new FeesAndCredits.Builder()
             .setCurrency(registry.getCurrency())
             .addFeeOrCredit(
-                Fee.create(getDomainRenewCost(domainName, date, 1).getAmount(), FeeType.RENEW))
+                Fee.create(
+                    domainPrices.getRenewCost().getAmount(),
+                    FeeType.RENEW,
+                    domainPrices.isPremium()))
             .addFeeOrCredit(
-                Fee.create(registry.getStandardRestoreCost().getAmount(), FeeType.RESTORE))
+                Fee.create(registry.getStandardRestoreCost().getAmount(), FeeType.RESTORE, false))
             .build();
     return customLogic.customizeRestorePrice(
         RestorePriceParameters.newBuilder()
             .setFeesAndCredits(feesAndCredits)
             .setRegistry(registry)
             .setDomainName(InternetDomainName.from(domainName))
-            .setAsOfDate(date)
+            .setAsOfDate(dateTime)
             .build());
   }
 
   /** Returns a new transfer price for the pricer. */
-  public FeesAndCredits getTransferPrice(Registry registry, String domainName, DateTime date)
+  public FeesAndCredits getTransferPrice(Registry registry, String domainName, DateTime dateTime)
       throws EppException {
-    Money renewCost = getDomainRenewCost(domainName, date, 1);
+    DomainPrices domainPrices = getPricesForDomainName(domainName, dateTime);
     return customLogic.customizeTransferPrice(
         TransferPriceParameters.newBuilder()
             .setFeesAndCredits(
                 new FeesAndCredits.Builder()
                     .setCurrency(registry.getCurrency())
-                    .addFeeOrCredit(Fee.create(renewCost.getAmount(), FeeType.RENEW))
+                    .addFeeOrCredit(
+                        Fee.create(
+                            domainPrices.getRenewCost().getAmount(),
+                            FeeType.RENEW,
+                            domainPrices.isPremium()))
                     .build())
             .setRegistry(registry)
             .setDomainName(InternetDomainName.from(domainName))
-            .setAsOfDate(date)
+            .setAsOfDate(dateTime)
             .build());
   }
 
   /** Returns a new update price for the pricer. */
-  public FeesAndCredits getUpdatePrice(Registry registry, String domainName, DateTime date)
+  public FeesAndCredits getUpdatePrice(Registry registry, String domainName, DateTime dateTime)
       throws EppException {
     CurrencyUnit currency = registry.getCurrency();
-    BaseFee feeOrCredit =
-        Fee.create(Money.zero(registry.getCurrency()).getAmount(), FeeType.UPDATE);
+    BaseFee feeOrCredit = Fee.create(zeroInCurrency(currency), FeeType.UPDATE, false);
     return customLogic.customizeUpdatePrice(
         UpdatePriceParameters.newBuilder()
             .setFeesAndCredits(
@@ -172,19 +182,19 @@ public final class DomainPricingLogic {
                     .build())
             .setRegistry(registry)
             .setDomainName(InternetDomainName.from(domainName))
-            .setAsOfDate(date)
+            .setAsOfDate(dateTime)
             .build());
   }
 
   /** Returns the fee class for a given domain and date. */
-  public Optional<String> getFeeClass(String domainName, DateTime date) {
-    return getDomainFeeClass(domainName, date);
+  public Optional<String> getFeeClass(String domainName, DateTime dateTime) {
+    return getDomainFeeClass(domainName, dateTime);
   }
 
+  /** Returns the domain create cost with allocation-token-related discounts applied. */
   private Money getDomainCreateCostWithDiscount(
-      String domainName, DateTime date, int years, Optional<AllocationToken> allocationToken)
+      DomainPrices domainPrices, int years, Optional<AllocationToken> allocationToken)
       throws EppException {
-    DomainPrices domainPrices = PricingEngineProxy.getPricesForDomainName(domainName, date);
     if (allocationToken.isPresent()
         && allocationToken.get().getDiscountFraction() != 0.0
         && domainPrices.isPremium()) {

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

@@ -211,8 +211,7 @@ public final class DomainRenewFlow implements TransactionalFlow {
             BeforeResponseParameters.newBuilder()
                 .setDomain(newDomain)
                 .setResData(DomainRenewData.create(targetId, newExpirationTime))
-                .setResponseExtensions(
-                    createResponseExtensions(feesAndCredits.getTotalCost(), feeRenew))
+                .setResponseExtensions(createResponseExtensions(feesAndCredits, feeRenew))
                 .build());
     return responseBuilder
         .setResData(responseData.resData())
@@ -270,14 +269,19 @@ public final class DomainRenewFlow implements TransactionalFlow {
   }
 
   private ImmutableList<FeeTransformResponseExtension> createResponseExtensions(
-      Money renewCost, Optional<FeeRenewCommandExtension> feeRenew) {
+      FeesAndCredits feesAndCredits, Optional<FeeRenewCommandExtension> feeRenew) {
     return feeRenew.isPresent()
         ? ImmutableList.of(
             feeRenew
                 .get()
                 .createResponseBuilder()
-                .setCurrency(renewCost.getCurrencyUnit())
-                .setFees(ImmutableList.of(Fee.create(renewCost.getAmount(), FeeType.RENEW)))
+                .setCurrency(feesAndCredits.getCurrency())
+                .setFees(
+                    ImmutableList.of(
+                        Fee.create(
+                            feesAndCredits.getRenewCost().getAmount(),
+                            FeeType.RENEW,
+                            feesAndCredits.hasPremiumFeesOfType(FeeType.RENEW))))
                 .build())
         : ImmutableList.of();
   }

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

@@ -143,24 +143,30 @@ public final class DomainRestoreRequestFlow implements TransactionalFlow  {
     verifyRestoreAllowed(command, existingDomain, feeUpdate, feesAndCredits, now);
     HistoryEntry historyEntry = buildHistoryEntry(existingDomain, now);
     ImmutableSet.Builder<ImmutableObject> entitiesToSave = new ImmutableSet.Builder<>();
-    entitiesToSave.addAll(
-        createRestoreAndRenewBillingEvents(
-            historyEntry, feesAndCredits.getRestoreCost(), feesAndCredits.getRenewCost(), now));
-    // We don't preserve the original expiration time of the domain when we restore, since doing so
-    // would require us to know if they received a grace period refund when they deleted the domain,
-    // and to charge them for that again. Instead, we just say that all restores get a fresh year of
-    // registration and bill them for that accordingly.
-    DateTime newExpirationTime = now.plusYears(1);
-    BillingEvent.Recurring autorenewEvent = newAutorenewBillingEvent(existingDomain)
-        .setEventTime(newExpirationTime)
-        .setRecurrenceEndTime(END_OF_TIME)
-        .setParent(historyEntry)
-        .build();
-    PollMessage.Autorenew autorenewPollMessage = newAutorenewPollMessage(existingDomain)
-        .setEventTime(newExpirationTime)
-        .setAutorenewEndTime(END_OF_TIME)
-        .setParent(historyEntry)
-        .build();
+
+    // Restore the expiration time on the deleted domain, except if that's already passed, then add
+    // a year and bill for it immediately, with no grace period.
+    DateTime newExpirationTime = existingDomain.getRegistrationExpirationTime();
+    if (newExpirationTime.isBefore(now)) {
+      entitiesToSave.add(createRenewBillingEvent(historyEntry, feesAndCredits.getRenewCost(), now));
+      newExpirationTime = newExpirationTime.plusYears(1);
+    }
+    // Always bill for the restore itself.
+    entitiesToSave.add(
+        createRestoreBillingEvent(historyEntry, feesAndCredits.getRestoreCost(), now));
+
+    BillingEvent.Recurring autorenewEvent =
+        newAutorenewBillingEvent(existingDomain)
+            .setEventTime(newExpirationTime)
+            .setRecurrenceEndTime(END_OF_TIME)
+            .setParent(historyEntry)
+            .build();
+    PollMessage.Autorenew autorenewPollMessage =
+        newAutorenewPollMessage(existingDomain)
+            .setEventTime(newExpirationTime)
+            .setAutorenewEndTime(END_OF_TIME)
+            .setParent(historyEntry)
+            .build();
     DomainBase newDomain =
         performRestore(
             existingDomain, newExpirationTime, autorenewEvent, autorenewPollMessage, now, clientId);
@@ -170,9 +176,7 @@ public final class DomainRestoreRequestFlow implements TransactionalFlow  {
     ofy().delete().key(existingDomain.getDeletePollMessage());
     dnsQueue.addDomainRefreshTask(existingDomain.getFullyQualifiedDomainName());
     return responseBuilder
-        .setExtensions(
-            createResponseExtensions(
-                feesAndCredits.getRestoreCost(), feesAndCredits.getRenewCost(), feeUpdate))
+        .setExtensions(createResponseExtensions(feesAndCredits, feeUpdate))
         .build();
   }
 
@@ -212,18 +216,6 @@ public final class DomainRestoreRequestFlow implements TransactionalFlow  {
     validateFeeChallenge(targetId, now, feeUpdate, feesAndCredits);
   }
 
-  private ImmutableSet<BillingEvent.OneTime> createRestoreAndRenewBillingEvents(
-      HistoryEntry historyEntry, Money restoreCost, Money renewCost, DateTime now) {
-    // Bill for the restore.
-    BillingEvent.OneTime restoreEvent = createRestoreBillingEvent(historyEntry, restoreCost, now);
-    // Create a new autorenew billing event and poll message starting at the new expiration time.
-    // Also bill for the 1 year cost of a domain renew. This is to avoid registrants being able to
-    // game the system for premium names by renewing, deleting, and then restoring to get a free
-    // year. Note that this billing event has no grace period; it is effective immediately.
-    BillingEvent.OneTime renewEvent = createRenewBillingEvent(historyEntry, renewCost, now);
-    return ImmutableSet.of(restoreEvent, renewEvent);
-  }
-
   private static DomainBase performRestore(
       DomainBase existingDomain,
       DateTime newExpirationTime,
@@ -271,17 +263,23 @@ public final class DomainRestoreRequestFlow implements TransactionalFlow  {
   }
 
   private static ImmutableList<FeeTransformResponseExtension> createResponseExtensions(
-      Money restoreCost, Money renewCost, Optional<FeeUpdateCommandExtension> feeUpdate) {
+      FeesAndCredits feesAndCredits, Optional<FeeUpdateCommandExtension> feeUpdate) {
     return feeUpdate.isPresent()
         ? ImmutableList.of(
             feeUpdate
                 .get()
                 .createResponseBuilder()
-                .setCurrency(restoreCost.getCurrencyUnit())
+                .setCurrency(feesAndCredits.getCurrency())
                 .setFees(
                     ImmutableList.of(
-                        Fee.create(restoreCost.getAmount(), FeeType.RESTORE),
-                        Fee.create(renewCost.getAmount(), FeeType.RENEW)))
+                        Fee.create(
+                            feesAndCredits.getRestoreCost().getAmount(),
+                            FeeType.RESTORE,
+                            feesAndCredits.hasPremiumFeesOfType(FeeType.RESTORE)),
+                        Fee.create(
+                            feesAndCredits.getRenewCost().getAmount(),
+                            FeeType.RENEW,
+                            feesAndCredits.hasPremiumFeesOfType(FeeType.RENEW))))
                 .build())
         : ImmutableList.of();
   }

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

@@ -15,6 +15,7 @@
 package google.registry.flows.domain;
 
 import static com.google.common.collect.ImmutableList.toImmutableList;
+import static google.registry.flows.domain.DomainFlowUtils.zeroInCurrency;
 import static google.registry.util.CollectionUtils.nullToEmpty;
 import static google.registry.util.CollectionUtils.nullToEmptyImmutableCopy;
 import static google.registry.util.PreconditionsUtils.checkArgumentNotNull;
@@ -27,6 +28,7 @@ import google.registry.model.domain.fee.BaseFee;
 import google.registry.model.domain.fee.BaseFee.FeeType;
 import google.registry.model.domain.fee.Credit;
 import google.registry.model.domain.fee.Fee;
+import java.math.BigDecimal;
 import org.joda.money.CurrencyUnit;
 import org.joda.money.Money;
 
@@ -39,26 +41,30 @@ public class FeesAndCredits extends ImmutableObject implements Buildable {
   private ImmutableList<Credit> credits;
 
   private Money getTotalCostForType(FeeType type) {
-    Money result = Money.zero(currency);
     checkArgumentNotNull(type);
-    for (Fee fee : fees) {
-      if (fee.getType() == type) {
-        result = result.plus(fee.getCost());
-      }
-    }
-    return result;
+    return Money.of(
+        currency,
+        fees.stream()
+            .filter(f -> f.getType() == type)
+            .map(BaseFee::getCost)
+            .reduce(zeroInCurrency(currency), BigDecimal::add));
+  }
+
+  public boolean hasPremiumFeesOfType(FeeType type) {
+    return fees.stream().filter(f -> f.getType() == type).anyMatch(BaseFee::isPremium);
   }
 
   /** Returns the total cost of all fees and credits for the event. */
   public Money getTotalCost() {
-    Money result = Money.zero(currency);
-    for (Fee fee : fees) {
-      result = result.plus(fee.getCost());
-    }
-    for (Credit credit : credits) {
-      result = result.plus(credit.getCost());
-    }
-    return result;
+    return Money.of(
+        currency,
+        Streams.concat(fees.stream(), credits.stream())
+            .map(BaseFee::getCost)
+            .reduce(zeroInCurrency(currency), BigDecimal::add));
+  }
+
+  public boolean hasAnyPremiumFees() {
+    return fees.stream().anyMatch(BaseFee::isPremium);
   }
 
   /** Returns the create cost for the event. */

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

@@ -35,9 +35,9 @@ import java.util.List;
 import org.joda.time.DateTime;
 
 /**
- * Shared entity for date cursors. This type supports both "scoped" cursors (i.e. per resource
- * of a given type, such as a TLD) and global (i.e. one per environment) cursors, defined internally
- * as scoped on {@link EntityGroupRoot}.
+ * Shared entity for date cursors. This type supports both "scoped" cursors (i.e. per resource of a
+ * given type, such as a TLD) and global (i.e. one per environment) cursors, defined internally as
+ * scoped on {@link EntityGroupRoot}.
  */
 @Entity
 public class Cursor extends ImmutableObject implements DatastoreEntity {

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

@@ -27,14 +27,14 @@ import google.registry.schema.replay.SqlEntity;
  * reasons.
  *
  * <p>This exists as a storage place for common configuration options and global settings that
- * aren't updated too frequently.  Entities in this entity group are usually cached upon load.  The
+ * aren't updated too frequently. Entities in this entity group are usually cached upon load. The
  * reason this common entity group exists is because it enables strongly consistent queries and
- * updates across this seldomly updated data.  This shared entity group also helps cut down on
- * a potential ballooning in the number of entity groups enlisted in transactions.
+ * updates across this seldomly updated data. This shared entity group also helps cut down on a
+ * potential ballooning in the number of entity groups enlisted in transactions.
  *
- * <p>Historically, each TLD used to have a separate namespace, and all entities for a TLD were in
- * a single EntityGroupRoot for that TLD.  Hence why there was a "cross-tld" entity group -- it was
- * the entity group for the single namespace where global data applicable for all TLDs lived.
+ * <p>Historically, each TLD used to have a separate namespace, and all entities for a TLD were in a
+ * single EntityGroupRoot for that TLD. Hence why there was a "cross-tld" entity group -- it was the
+ * entity group for the single namespace where global data applicable for all TLDs lived.
  */
 @Entity
 public class EntityGroupRoot extends BackupGroupRoot implements DatastoreEntity {

core/src/main/java/google/registry/model/domain/fee/BaseFee.java

@@ -104,6 +104,8 @@ public abstract class BaseFee extends ImmutableObject {
 
   @XmlTransient Range<DateTime> validDateRange;
 
+  @XmlTransient boolean isPremium;
+
   public String getDescription() {
     return description;
   }
@@ -120,6 +122,11 @@ public abstract class BaseFee extends ImmutableObject {
     return firstNonNull(refundable, true);
   }
 
+  /** Returns whether the fee in question is a premium price. */
+  public boolean isPremium() {
+    return isPremium;
+  }
+
   /**
    * According to the fee extension specification, a fee must always be non-negative, while a credit
    * must always be negative. Essentially, they are the same thing, just with different sign.

core/src/main/java/google/registry/model/domain/fee/Fee.java

@@ -31,25 +31,33 @@ import org.joda.time.DateTime;
 public class Fee extends BaseFee {
 
   /** Creates a Fee for the given cost and type with the default description. */
-  public static Fee create(BigDecimal cost, FeeType type, Object... descriptionArgs) {
+  public static Fee create(
+      BigDecimal cost, FeeType type, boolean isPremium, Object... descriptionArgs) {
     checkArgumentNotNull(type, "Must specify the type of the fee");
-    return createWithCustomDescription(cost, type, type.renderDescription(descriptionArgs));
+    return createWithCustomDescription(
+        cost, type, isPremium, type.renderDescription(descriptionArgs));
   }
 
   /** Creates a Fee for the given cost, type, and valid date range with the default description. */
   public static Fee create(
-      BigDecimal cost, FeeType type, Range<DateTime> validDateRange, Object... descriptionArgs) {
-    Fee instance = create(cost, type, descriptionArgs);
+      BigDecimal cost,
+      FeeType type,
+      boolean isPremium,
+      Range<DateTime> validDateRange,
+      Object... descriptionArgs) {
+    Fee instance = create(cost, type, isPremium, descriptionArgs);
     instance.validDateRange = validDateRange;
     return instance;
   }
 
   /** Creates a Fee for the given cost and type with a custom description. */
-  public static Fee createWithCustomDescription(BigDecimal cost, FeeType type, String description) {
+  private static Fee createWithCustomDescription(
+      BigDecimal cost, FeeType type, boolean isPremium, String description) {
     Fee instance = new Fee();
     instance.cost = checkNotNull(cost);
-    checkArgument(instance.cost.signum() >= 0);
+    checkArgument(instance.cost.signum() >= 0, "Cost must be a positive number");
     instance.type = checkNotNull(type);
+    instance.isPremium = isPremium;
     instance.description = description;
     return instance;
   }

core/src/main/java/google/registry/model/domain/fee11/FeeCheckCommandExtensionV11.java

@@ -56,7 +56,7 @@ public class FeeCheckCommandExtensionV11 extends ImmutableObject
   /** The period to check. */
   Period period;
 
-  /** The class to check. */
+  /** The fee class to check. */
   @XmlElement(name = "class")
   String feeClass;
 

core/src/main/java/google/registry/model/ofy/CommitLogBucket.java

@@ -44,10 +44,9 @@ import org.joda.time.DateTime;
 /**
  * Root for a random commit log bucket.
  *
- * <p>This is used to shard {@link CommitLogManifest} objects into
- * {@link RegistryConfig#getCommitLogBucketCount() N} entity groups. This increases
- * transaction throughput, while maintaining the ability to perform strongly-consistent ancestor
- * queries.
+ * <p>This is used to shard {@link CommitLogManifest} objects into {@link
+ * RegistryConfig#getCommitLogBucketCount() N} entity groups. This increases transaction throughput,
+ * while maintaining the ability to perform strongly-consistent ancestor queries.
  *
  * @see <a href="https://cloud.google.com/appengine/articles/scaling/contention">Avoiding Datastore
  *     contention</a>

core/src/main/java/google/registry/model/ofy/CommitLogCheckpoint.java

@@ -38,11 +38,11 @@ import org.joda.time.DateTime;
  * Entity representing a point-in-time consistent view of Datastore, based on commit logs.
  *
  * <p>Conceptually, this entity consists of two pieces of information: the checkpoint "wall" time
- * and a set of bucket checkpoint times.  The former is the ID for this checkpoint (constrained
- * to be unique upon checkpoint creation) and also represents the approximate wall time of the
- * consistent Datastore view this checkpoint represents.  The latter is really a mapping from
- * bucket ID to timestamp, where the timestamp dictates the upper bound (inclusive) on commit logs
- * from that bucket to include when restoring Datastore to this checkpoint.
+ * and a set of bucket checkpoint times. The former is the ID for this checkpoint (constrained to be
+ * unique upon checkpoint creation) and also represents the approximate wall time of the consistent
+ * Datastore view this checkpoint represents. The latter is really a mapping from bucket ID to
+ * timestamp, where the timestamp dictates the upper bound (inclusive) on commit logs from that
+ * bucket to include when restoring Datastore to this checkpoint.
  */
 @Entity
 @NotBackedUp(reason = Reason.COMMIT_LOGS)

core/src/main/java/google/registry/model/ofy/CommitLogCheckpointRoot.java

@@ -28,9 +28,7 @@ import google.registry.schema.replay.DatastoreEntity;
 import google.registry.schema.replay.SqlEntity;
 import org.joda.time.DateTime;
 
-/**
- * Singleton parent entity for all commit log checkpoints.
- */
+/** Singleton parent entity for all commit log checkpoints. */
 @Entity
 @NotBackedUp(reason = Reason.COMMIT_LOGS)
 public class CommitLogCheckpointRoot extends ImmutableObject implements DatastoreEntity {

core/src/main/java/google/registry/model/registry/Registry.java

@@ -579,6 +579,8 @@ public class Registry extends ImmutableObject implements Buildable {
     return Fee.create(
         eapFeeSchedule.getValueAtTime(now).getAmount(),
         FeeType.EAP,
+        // An EAP fee counts as premium so the domain's overall Fee doesn't show as standard-priced.
+        true,
         validPeriod,
         validPeriod.upperEndpoint());
   }

core/src/main/java/google/registry/model/registry/label/PremiumList.java

@@ -248,7 +248,7 @@ public final class PremiumList extends BaseDomainLabelList<Money, PremiumList.Pr
   }
 
   /**
-   * A premium list entry entity, persisted to Datastore.  Each instance represents the price of a
+   * A premium list entry entity, persisted to Datastore. Each instance represents the price of a
    * single label on a given TLD.
    */
   @ReportedOn

core/src/main/java/google/registry/model/registry/label/ReservedList.java

@@ -62,8 +62,8 @@ import org.joda.time.DateTime;
  */
 @Entity
 public final class ReservedList
-    extends BaseDomainLabelList<ReservationType, ReservedList.ReservedListEntry> implements
-    DatastoreEntity {
+    extends BaseDomainLabelList<ReservationType, ReservedList.ReservedListEntry>
+    implements DatastoreEntity {
 
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 

core/src/main/java/google/registry/model/tmch/ClaimsListShard.java

@@ -327,8 +327,8 @@ public class ClaimsListShard extends ImmutableObject implements DatastoreEntity
   }
 
   /**
-   * Serves as the coordinating claims list singleton linking to the {@link ClaimsListRevision}
-   * that is live.
+   * Serves as the coordinating claims list singleton linking to the {@link ClaimsListRevision} that
+   * is live.
    */
   @Entity
   @NotBackedUp(reason = Reason.EXTERNALLY_SOURCED)

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

@@ -88,8 +88,8 @@ public final class LevelDbLogReader implements Iterator<byte[]> {
   }
 
   /**
-   * Returns the next {@link #BLOCK_SIZE} bytes from the input channel, or {@link
-   * Optional#empty()} if there is no more data.
+   * Returns the next {@link #BLOCK_SIZE} bytes from the input channel, or {@link Optional#empty()}
+   * if there is no more data.
    */
   // TODO(weiminyu): use ByteBuffer directly.
   private Optional<byte[]> readFromChannel() throws IOException {

core/src/main/java/google/registry/ui/server/registrar/RegistryLockPostAction.java

@@ -30,7 +30,6 @@ import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
 import com.google.common.flogger.FluentLogger;
 import com.google.gson.Gson;
-import google.registry.config.RegistryConfig;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.model.registrar.Registrar;
 import google.registry.model.registrar.RegistrarContact;
@@ -49,12 +48,12 @@ import google.registry.tools.DomainLockUtils;
 import google.registry.util.EmailMessage;
 import google.registry.util.SendEmailService;
 import java.net.URISyntaxException;
-import java.net.URL;
 import java.util.Map;
 import java.util.Optional;
 import javax.inject.Inject;
 import javax.mail.internet.AddressException;
 import javax.mail.internet.InternetAddress;
+import javax.servlet.http.HttpServletRequest;
 import org.apache.http.client.utils.URIBuilder;
 import org.joda.time.Duration;
 
@@ -76,11 +75,11 @@ public class RegistryLockPostAction implements Runnable, JsonActionRunner.JsonAc
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
   private static final Gson GSON = new Gson();
 
-  private static final URL URL_BASE = RegistryConfig.getDefaultServer();
   private static final String VERIFICATION_EMAIL_TEMPLATE =
       "Please click the link below to perform the lock / unlock action on domain %s. Note: "
           + "this code will expire in one hour.\n\n%s";
 
+  private final HttpServletRequest req;
   private final JsonActionRunner jsonActionRunner;
   private final AuthResult authResult;
   private final AuthenticatedRegistrarAccessor registrarAccessor;
@@ -90,12 +89,14 @@ public class RegistryLockPostAction implements Runnable, JsonActionRunner.JsonAc
 
   @Inject
   RegistryLockPostAction(
+      HttpServletRequest req,
       JsonActionRunner jsonActionRunner,
       AuthResult authResult,
       AuthenticatedRegistrarAccessor registrarAccessor,
       SendEmailService sendEmailService,
       DomainLockUtils domainLockUtils,
       @Config("gSuiteOutgoingEmailAddress") InternetAddress gSuiteOutgoingEmailAddress) {
+    this.req = req;
     this.jsonActionRunner = jsonActionRunner;
     this.authResult = authResult;
     this.registrarAccessor = registrarAccessor;
@@ -161,7 +162,7 @@ public class RegistryLockPostAction implements Runnable, JsonActionRunner.JsonAc
       String url =
           new URIBuilder()
               .setScheme("https")
-              .setHost(URL_BASE.getHost())
+              .setHost(req.getServerName())
               .setPath("registry-lock-verify")
               .setParameter("lockVerificationCode", lock.getVerificationCode())
               .setParameter("isLock", String.valueOf(isLock))

core/src/test/java/google/registry/backup/CommitLogExports.java

@@ -0,0 +1,231 @@
+// Copyright 2017 The Nomulus Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package google.registry.backup;
+
+import static com.google.common.base.MoreObjects.firstNonNull;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.collect.ImmutableList.toImmutableList;
+import static com.google.common.collect.Iterables.concat;
+import static com.google.common.collect.Lists.partition;
+import static google.registry.backup.BackupUtils.serializeEntity;
+import static google.registry.model.ofy.CommitLogBucket.getBucketKey;
+import static google.registry.model.ofy.ObjectifyService.ofy;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.util.DateTimeUtils.START_OF_TIME;
+import static google.registry.util.DateTimeUtils.isAtOrAfter;
+import static java.util.Comparator.comparingLong;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Streams;
+import com.googlecode.objectify.Key;
+import google.registry.model.ImmutableObject;
+import google.registry.model.ofy.CommitLogBucket;
+import google.registry.model.ofy.CommitLogCheckpoint;
+import google.registry.model.ofy.CommitLogCheckpointRoot;
+import google.registry.model.ofy.CommitLogManifest;
+import google.registry.model.ofy.CommitLogMutation;
+import google.registry.util.Clock;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import javax.annotation.Nullable;
+import org.joda.time.DateTime;
+
+/**
+ * Helpers for exporting the diff between two commit log checkpoints to a local file.
+ *
+ * <p>In production, CommitLogs are saved periodically by cron jobs. During each job, the {@link
+ * CommitLogCheckpointAction} is invoked first to compute a {@link CommitLogCheckpoint} and persist
+ * it in Datastore. Then the {@link ExportCommitLogDiffAction} is invoked to export the diffs
+ * accumulated between the previous and current checkpoints to a file.
+ *
+ * <p>The {@link #computeCheckpoint(Clock)} method is copied with simplification from {@link
+ * CommitLogCheckpointAction}, and the {@link #saveCommitLogs(String, CommitLogCheckpoint,
+ * CommitLogCheckpoint)} method is copied with simplification from {@link
+ * ExportCommitLogDiffAction}. We opted for copying instead of refactoring to reduce risk to
+ * production code.
+ */
+public final class CommitLogExports {
+
+  public static final String DIFF_FILE_PREFIX = "commit_diff_until_";
+
+  private static final int EXPORT_DIFF_BATCH_SIZE = 100;
+
+  private CommitLogExports() {}
+
+  /**
+   * Returns the next {@link CommitLogCheckpoint} for Commit logs. Please refer to the class javadoc
+   * for background.
+   */
+  public static CommitLogCheckpoint computeCheckpoint(Clock clock) {
+    CommitLogCheckpointStrategy strategy = new CommitLogCheckpointStrategy();
+    strategy.clock = clock;
+    strategy.ofy = ofy();
+
+    CommitLogCheckpoint checkpoint = strategy.computeCheckpoint();
+    tm().transact(
+            () -> {
+              DateTime lastWrittenTime = CommitLogCheckpointRoot.loadRoot().getLastWrittenTime();
+              checkState(
+                  checkpoint.getCheckpointTime().isAfter(lastWrittenTime),
+                  "Newer checkpoint already written at time: %s",
+                  lastWrittenTime);
+              ofy()
+                  .saveWithoutBackup()
+                  .entities(
+                      checkpoint, CommitLogCheckpointRoot.create(checkpoint.getCheckpointTime()));
+            });
+    return checkpoint;
+  }
+
+  /**
+   * Saves the incremental changes between {@code prevCheckpoint} and {@code checkpoint} and returns
+   * the {@link File}. Please refer to class javadoc for background.
+   */
+  public static File saveCommitLogs(
+      String commitLogDir,
+      @Nullable CommitLogCheckpoint prevCheckpoint,
+      CommitLogCheckpoint checkpoint) {
+    checkArgument(
+        prevCheckpoint == null
+            || (isAtOrAfter(prevCheckpoint.getCheckpointTime(), START_OF_TIME)
+                && prevCheckpoint.getCheckpointTime().isBefore(checkpoint.getCheckpointTime())),
+        "Inversed checkpoint: prev is %s, current is %s.",
+        Optional.ofNullable(prevCheckpoint)
+            .map(CommitLogCheckpoint::getCheckpointTime)
+            .map(DateTime::toString)
+            .orElse("null"),
+        checkpoint.getCheckpointTime().toString());
+
+    // Load the keys of all the manifests to include in this diff.
+    List<Key<CommitLogManifest>> sortedKeys = loadAllDiffKeys(prevCheckpoint, checkpoint);
+    // Open an output channel to GCS, wrapped in a stream for convenience.
+    File commitLogFile =
+        new File(commitLogDir + "/" + DIFF_FILE_PREFIX + checkpoint.getCheckpointTime());
+    try (OutputStream commitLogStream =
+        new BufferedOutputStream(new FileOutputStream(commitLogFile))) {
+      // Export the upper checkpoint itself.
+      serializeEntity(checkpoint, commitLogStream);
+      // If there are no manifests to export, stop early, now that we've written out the file with
+      // the checkpoint itself (which is needed for restores, even if it's empty).
+      if (sortedKeys.isEmpty()) {
+        return commitLogFile;
+      }
+      // Export to GCS in chunks, one per fixed batch of commit logs.  While processing one batch,
+      // asynchronously load the entities for the next one.
+      List<List<Key<CommitLogManifest>>> keyChunks = partition(sortedKeys, EXPORT_DIFF_BATCH_SIZE);
+      // Objectify's map return type is asynchronous. Calling .values() will block until it loads.
+      Map<?, CommitLogManifest> nextChunkToExport = ofy().load().keys(keyChunks.get(0));
+      for (int i = 0; i < keyChunks.size(); i++) {
+        // Force the async load to finish.
+        Collection<CommitLogManifest> chunkValues = nextChunkToExport.values();
+        // Since there is no hard bound on how much data this might be, take care not to let the
+        // Objectify session cache fill up and potentially run out of memory. This is the only safe
+        // point to do this since at this point there is no async load in progress.
+        ofy().clearSessionCache();
+        // Kick off the next async load, which can happen in parallel to the current GCS export.
+        if (i + 1 < keyChunks.size()) {
+          nextChunkToExport = ofy().load().keys(keyChunks.get(i + 1));
+        }
+        exportChunk(commitLogStream, chunkValues);
+      }
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+    return commitLogFile;
+  }
+
+  /**
+   * Loads all the diff keys, sorted in a transaction-consistent chronological order.
+   *
+   * @param lowerCheckpoint exclusive lower bound on keys in this diff, or null if no lower bound
+   * @param upperCheckpoint inclusive upper bound on keys in this diff
+   */
+  private static ImmutableList<Key<CommitLogManifest>> loadAllDiffKeys(
+      @Nullable final CommitLogCheckpoint lowerCheckpoint,
+      final CommitLogCheckpoint upperCheckpoint) {
+    // Fetch the keys (no data) between these checkpoints, and sort by timestamp. This ordering is
+    // transaction-consistent by virtue of our checkpoint strategy and our customized Ofy; see
+    // CommitLogCheckpointStrategy for the proof. We break ties by sorting on bucket ID to ensure
+    // a deterministic order.
+    return upperCheckpoint.getBucketTimestamps().keySet().stream()
+        .flatMap(
+            bucketNum ->
+                Streams.stream(loadDiffKeysFromBucket(lowerCheckpoint, upperCheckpoint, bucketNum)))
+        .sorted(
+            comparingLong(Key<CommitLogManifest>::getId)
+                .thenComparingLong(a -> a.getParent().getId()))
+        .collect(toImmutableList());
+  }
+
+  /**
+   * Loads the diff keys for one bucket.
+   *
+   * @param lowerCheckpoint exclusive lower bound on keys in this diff, or null if no lower bound
+   * @param upperCheckpoint inclusive upper bound on keys in this diff
+   * @param bucketNum the bucket to load diff keys from
+   */
+  private static Iterable<Key<CommitLogManifest>> loadDiffKeysFromBucket(
+      @Nullable CommitLogCheckpoint lowerCheckpoint,
+      CommitLogCheckpoint upperCheckpoint,
+      int bucketNum) {
+    // If no lower checkpoint exists, or if it exists but had no timestamp for this bucket number
+    // (because the bucket count was increased between these checkpoints), then use START_OF_TIME
+    // as the effective exclusive lower bound.
+    DateTime lowerCheckpointBucketTime =
+        firstNonNull(
+            (lowerCheckpoint == null) ? null : lowerCheckpoint.getBucketTimestamps().get(bucketNum),
+            START_OF_TIME);
+    // Since START_OF_TIME=0 is not a valid id in a key, add 1 to both bounds. Then instead of
+    // loading lowerBound < x <= upperBound, we can load lowerBound <= x < upperBound.
+    DateTime lowerBound = lowerCheckpointBucketTime.plusMillis(1);
+    DateTime upperBound = upperCheckpoint.getBucketTimestamps().get(bucketNum).plusMillis(1);
+    // If the lower and upper bounds are equal, there can't be any results, so skip the query.
+    if (lowerBound.equals(upperBound)) {
+      return ImmutableSet.of();
+    }
+    Key<CommitLogBucket> bucketKey = getBucketKey(bucketNum);
+    return ofy()
+        .load()
+        .type(CommitLogManifest.class)
+        .ancestor(bucketKey)
+        .filterKey(">=", CommitLogManifest.createKey(bucketKey, lowerBound))
+        .filterKey("<", CommitLogManifest.createKey(bucketKey, upperBound))
+        .keys();
+  }
+
+  /** Writes a chunks-worth of manifests and associated mutations to GCS. */
+  private static void exportChunk(OutputStream gcsStream, Collection<CommitLogManifest> chunk)
+      throws IOException {
+    // Kickoff async loads for all the manifests in the chunk.
+    ImmutableList.Builder<Iterable<? extends ImmutableObject>> entities =
+        new ImmutableList.Builder<>();
+    for (CommitLogManifest manifest : chunk) {
+      entities.add(ImmutableList.of(manifest));
+      entities.add(ofy().load().type(CommitLogMutation.class).ancestor(manifest));
+    }
+    for (ImmutableObject entity : concat(entities.build())) {
+      serializeEntity(entity, gcsStream);
+    }
+  }
+}

core/src/test/java/google/registry/beam/initsql/BackupPathsTest.java

@@ -1,50 +0,0 @@
-// Copyright 2020 The Nomulus Authors. All Rights Reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package google.registry.beam.initsql;
-
-import static com.google.common.truth.Truth.assertThat;
-import static google.registry.beam.initsql.BackupPaths.getKindFromFileName;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-import org.junit.jupiter.api.Test;
-
-/** Unit tests for {@link google.registry.beam.initsql.BackupPaths}. */
-public class BackupPathsTest {
-
-  @Test
-  void getKindFromFileName_empty() {
-    assertThrows(IllegalArgumentException.class, () -> getKindFromFileName(""));
-  }
-
-  @Test
-  void getKindFromFileName_notMatch() {
-    assertThrows(
-        IllegalArgumentException.class,
-        () -> getKindFromFileName("/tmp/all_namespaces/kind_/input-0"));
-  }
-
-  @Test
-  void getKindFromFileName_success() {
-    assertThat(getKindFromFileName("scheme:/somepath/all_namespaces/kind_mykind/input-something"))
-        .isEqualTo("mykind");
-  }
-
-  @Test
-  void getKindFromFileName_specialChar_success() {
-    assertThat(
-            getKindFromFileName("scheme:/somepath/all_namespaces/kind_.*+? /(a)/input-something"))
-        .isEqualTo(".*+? /(a)");
-  }
-}

core/src/test/java/google/registry/beam/initsql/BackupTestStore.java

@@ -20,6 +20,8 @@ import static google.registry.persistence.transaction.TransactionManagerFactory.
 
 import com.google.appengine.api.datastore.Entity;
 import com.googlecode.objectify.Key;
+import google.registry.backup.CommitLogExports;
+import google.registry.model.ofy.CommitLogCheckpoint;
 import google.registry.testing.AppEngineRule;
 import google.registry.testing.FakeClock;
 import google.registry.tools.LevelDbFileBuilder;
@@ -35,6 +37,10 @@ import org.joda.time.format.DateTimeFormatter;
  * <p>A Datastore backup consists of an unsynchronized data export and a sequence of incremental
  * Commit Logs that overlap with the export process. Together they can be used to recreate a
  * consistent snapshot of the Datastore.
+ *
+ * <p>For convenience of test-writing, the {@link #fakeClock} is advanced by 1 millisecond after
+ * every transaction is invoked on this store, ensuring strictly increasing timestamps on causally
+ * dependent transactions. In production, the same ordering is ensured by sleep and retry.
  */
 class BackupTestStore implements AutoCloseable {
 
@@ -44,6 +50,8 @@ class BackupTestStore implements AutoCloseable {
   private final FakeClock fakeClock;
   private AppEngineRule appEngine;
 
+  private CommitLogCheckpoint prevCommitLogCheckpoint;
+
   BackupTestStore(FakeClock fakeClock) throws Exception {
     this.fakeClock = fakeClock;
     this.appEngine =
@@ -55,16 +63,27 @@ class BackupTestStore implements AutoCloseable {
     this.appEngine.beforeEach(null);
   }
 
+  void transact(Iterable<Object> deletes, Iterable<Object> newOrUpdated) {
+    tm().transact(
+            () -> {
+              ofy().delete().entities(deletes);
+              ofy().save().entities(newOrUpdated);
+            });
+    fakeClock.advanceOneMilli();
+  }
+
   /** Inserts or updates {@code entities} in the Datastore. */
   @SafeVarargs
   final void insertOrUpdate(Object... entities) {
     tm().transact(() -> ofy().save().entities(entities).now());
+    fakeClock.advanceOneMilli();
   }
 
   /** Deletes {@code entities} from the Datastore. */
   @SafeVarargs
   final void delete(Object... entities) {
     tm().transact(() -> ofy().delete().entities(entities).now());
+    fakeClock.advanceOneMilli();
   }
 
   /**
@@ -109,8 +128,12 @@ class BackupTestStore implements AutoCloseable {
     builder.build();
   }
 
-  void saveCommitLog() {
-    throw new UnsupportedOperationException("Not implemented yet");
+  File saveCommitLogs(String commitLogDir) {
+    CommitLogCheckpoint checkpoint = CommitLogExports.computeCheckpoint(fakeClock);
+    File commitLogFile =
+        CommitLogExports.saveCommitLogs(commitLogDir, prevCommitLogCheckpoint, checkpoint);
+    prevCommitLogCheckpoint = checkpoint;
+    return commitLogFile;
   }
 
   @Override

core/src/test/java/google/registry/beam/initsql/BackupTestStoreTest.java

@@ -19,27 +19,36 @@ import static com.google.common.truth.Truth.assertWithMessage;
 import static com.google.common.truth.Truth8.assertThat;
 import static google.registry.model.common.EntityGroupRoot.getCrossTldKey;
 import static google.registry.model.ofy.ObjectifyService.ofy;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.testing.DatastoreHelper.newContactResource;
 import static google.registry.testing.DatastoreHelper.newDomainBase;
 import static google.registry.testing.DatastoreHelper.newRegistry;
 
 import com.google.appengine.api.datastore.Entity;
 import com.google.appengine.api.datastore.EntityTranslator;
+import com.google.common.base.Predicates;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Streams;
 import com.google.storage.onestore.v3.OnestoreEntity.EntityProto;
 import com.googlecode.objectify.Key;
+import google.registry.backup.CommitLogImports;
+import google.registry.backup.VersionedEntity;
 import google.registry.model.contact.ContactResource;
+import google.registry.model.domain.DesignatedContact;
 import google.registry.model.domain.DomainBase;
+import google.registry.model.ofy.Ofy;
 import google.registry.model.registry.Registry;
+import google.registry.persistence.VKey;
 import google.registry.testing.FakeClock;
+import google.registry.testing.InjectRule;
 import google.registry.tools.LevelDbLogReader;
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.Collections;
+import java.util.Optional;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Stream;
@@ -47,6 +56,7 @@ import org.joda.time.DateTime;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.api.io.TempDir;
 
 /** Unit tests for {@link BackupTestStore}. */
@@ -56,17 +66,25 @@ public class BackupTestStoreTest {
   private FakeClock fakeClock;
   private BackupTestStore store;
 
+  private Registry registry;
+  private ContactResource contact;
+  private DomainBase domain;
+
   @TempDir File tempDir;
 
+  @RegisterExtension InjectRule injectRule = new InjectRule();
+
   @BeforeEach
   void beforeEach() throws Exception {
     fakeClock = new FakeClock(START_TIME);
     store = new BackupTestStore(fakeClock);
+    injectRule.setStaticField(Ofy.class, "clock", fakeClock);
 
-    store.insertOrUpdate(newRegistry("tld1", "TLD1"));
-    ContactResource contact1 = newContactResource("contact_1");
-    DomainBase domain1 = newDomainBase("domain1.tld1", contact1);
-    store.insertOrUpdate(contact1, domain1);
+    registry = newRegistry("tld1", "TLD1");
+    store.insertOrUpdate(registry);
+    contact = newContactResource("contact_1");
+    domain = newDomainBase("domain1.tld1", contact);
+    store.insertOrUpdate(contact, domain);
   }
 
   @AfterEach
@@ -77,7 +95,8 @@ public class BackupTestStoreTest {
   @Test
   void export_filesCreated() throws IOException {
     String exportRootPath = tempDir.getAbsolutePath();
-    File exportFolder = new File(exportRootPath, "2000-01-01T00:00:00_000");
+    assertThat(fakeClock.nowUtc().toString()).isEqualTo("2000-01-01T00:00:00.002Z");
+    File exportFolder = new File(exportRootPath, "2000-01-01T00:00:00_002");
     assertWithMessage("Directory %s should not exist.", exportFolder.getAbsoluteFile())
         .that(exportFolder.exists())
         .isFalse();
@@ -100,7 +119,7 @@ public class BackupTestStoreTest {
   void export_folderNameChangesWithTime() throws IOException {
     String exportRootPath = tempDir.getAbsolutePath();
     fakeClock.advanceOneMilli();
-    File exportFolder = new File(exportRootPath, "2000-01-01T00:00:00_001");
+    File exportFolder = new File(exportRootPath, "2000-01-01T00:00:00_003");
     assertWithMessage("Directory %s should not exist.", exportFolder.getAbsoluteFile())
         .that(exportFolder.exists())
         .isFalse();
@@ -147,6 +166,69 @@ public class BackupTestStoreTest {
     assertThat(tlds).containsExactly("tld2");
   }
 
+  @Test
+  void saveCommitLogs_fileCreated() {
+    File commitLogFile = store.saveCommitLogs(tempDir.getAbsolutePath());
+    assertThat(commitLogFile.exists()).isTrue();
+    assertThat(commitLogFile.getName()).isEqualTo("commit_diff_until_2000-01-01T00:00:00.002Z");
+  }
+
+  @Test
+  void saveCommitLogs_inserts() {
+    File commitLogFile = store.saveCommitLogs(tempDir.getAbsolutePath());
+    assertThat(commitLogFile.exists()).isTrue();
+    ImmutableList<VersionedEntity> mutations = CommitLogImports.loadEntities(commitLogFile);
+    assertThat(mutations.stream().map(VersionedEntity::getEntity).map(Optional::get))
+        .containsExactlyElementsIn(toDatastoreEntities(registry, contact, domain));
+    // Registry created at -2, contract and domain created at -1.
+    assertThat(mutations.stream().map(VersionedEntity::commitTimeMills))
+        .containsExactly(
+            fakeClock.nowUtc().getMillis() - 2,
+            fakeClock.nowUtc().getMillis() - 1,
+            fakeClock.nowUtc().getMillis() - 1);
+  }
+
+  @Test
+  void saveCommitLogs_deletes() {
+    fakeClock.advanceOneMilli();
+    store.saveCommitLogs(tempDir.getAbsolutePath());
+    ContactResource newContact = newContactResource("contact2");
+    VKey<ContactResource> vKey = newContact.createVKey();
+    domain =
+        domain
+            .asBuilder()
+            .setRegistrant(vKey)
+            .setContacts(
+                ImmutableSet.of(
+                    DesignatedContact.create(DesignatedContact.Type.ADMIN, vKey),
+                    DesignatedContact.create(DesignatedContact.Type.TECH, vKey)))
+            .build();
+    store.insertOrUpdate(domain, newContact);
+    store.delete(contact);
+    fakeClock.advanceOneMilli();
+    File commitLogFile = store.saveCommitLogs(tempDir.getAbsolutePath());
+    ImmutableList<VersionedEntity> mutations = CommitLogImports.loadEntities(commitLogFile);
+    assertThat(mutations.stream().filter(VersionedEntity::isDelete).map(VersionedEntity::key))
+        .containsExactly(Key.create(contact).getRaw());
+
+    assertThat(
+            mutations.stream()
+                .filter(Predicates.not(VersionedEntity::isDelete))
+                .map(VersionedEntity::getEntity)
+                .map(Optional::get))
+        .containsExactlyElementsIn(toDatastoreEntities(domain, newContact));
+  }
+
+  @Test
+  void saveCommitLogs_empty() {
+    fakeClock.advanceOneMilli();
+    store.saveCommitLogs(tempDir.getAbsolutePath());
+    fakeClock.advanceOneMilli();
+    File commitLogFile = store.saveCommitLogs(tempDir.getAbsolutePath());
+    assertThat(commitLogFile.exists()).isTrue();
+    assertThat(CommitLogImports.loadEntities(commitLogFile)).isEmpty();
+  }
+
   private File export(String exportRootPath, Set<Key<?>> excludes) throws IOException {
     return store.export(
         exportRootPath,
@@ -168,4 +250,13 @@ public class BackupTestStoreTest {
     Entity entity = EntityTranslator.createFromPb(proto);
     return ofyEntityType.cast(ofy().load().fromEntity(entity));
   }
+
+  @SafeVarargs
+  private static ImmutableList<Entity> toDatastoreEntities(Object... ofyEntities) {
+    return tm().transact(
+            () ->
+                Stream.of(ofyEntities)
+                    .map(oe -> ofy().save().toEntity(oe))
+                    .collect(ImmutableList.toImmutableList()));
+  }
 }

core/src/test/java/google/registry/beam/initsql/CommitLogTransformsTest.java

@@ -0,0 +1,223 @@
+// Copyright 2020 The Nomulus Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package google.registry.beam.initsql;
+
+import static google.registry.model.ofy.ObjectifyService.ofy;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
+import static google.registry.testing.DatastoreHelper.newContactResource;
+import static google.registry.testing.DatastoreHelper.newDomainBase;
+import static google.registry.testing.DatastoreHelper.newRegistry;
+
+import com.google.appengine.api.datastore.Entity;
+import com.google.common.collect.ImmutableList;
+import google.registry.backup.VersionedEntity;
+import google.registry.model.contact.ContactResource;
+import google.registry.model.domain.DomainBase;
+import google.registry.model.ofy.Ofy;
+import google.registry.model.registry.Registry;
+import google.registry.testing.FakeClock;
+import google.registry.testing.InjectRule;
+import java.io.File;
+import java.io.Serializable;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.fs.MatchResult.Metadata;
+import org.apache.beam.sdk.testing.NeedsRunner;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.DateTime;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Unit tests for {@link CommitLogTransforms}. */
+// TODO(weiminyu): Upgrade to JUnit5 when TestPipeline is upgraded. It is also easy to adapt with
+// a wrapper.
+@RunWith(JUnit4.class)
+public class CommitLogTransformsTest implements Serializable {
+  private static final DateTime START_TIME = DateTime.parse("2000-01-01T00:00:00.0Z");
+
+  @Rule public final transient TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+  @Rule public final transient InjectRule injectRule = new InjectRule();
+
+  @Rule
+  public final transient TestPipeline pipeline =
+      TestPipeline.create().enableAbandonedNodeEnforcement(true);
+
+  private FakeClock fakeClock;
+  private transient BackupTestStore store;
+  private File commitLogsDir;
+  private File firstCommitLogFile;
+  // Canned data that are persisted to Datastore, used by assertions in tests.
+  // TODO(weiminyu): use Ofy entity pojos directly.
+  private transient ImmutableList<Entity> persistedEntities;
+
+  @Before
+  public void beforeEach() throws Exception {
+    fakeClock = new FakeClock(START_TIME);
+    store = new BackupTestStore(fakeClock);
+    injectRule.setStaticField(Ofy.class, "clock", fakeClock);
+
+    Registry registry = newRegistry("tld1", "TLD1");
+    store.insertOrUpdate(registry);
+    ContactResource contact1 = newContactResource("contact_1");
+    DomainBase domain1 = newDomainBase("domain1.tld1", contact1);
+    store.insertOrUpdate(contact1, domain1);
+    persistedEntities =
+        ImmutableList.of(registry, contact1, domain1).stream()
+            .map(ofyEntity -> tm().transact(() -> ofy().save().toEntity(ofyEntity)))
+            .collect(ImmutableList.toImmutableList());
+    commitLogsDir = temporaryFolder.newFolder();
+    firstCommitLogFile = store.saveCommitLogs(commitLogsDir.getAbsolutePath());
+  }
+
+  @After
+  public void afterEach() throws Exception {
+    if (store != null) {
+      store.close();
+      store = null;
+    }
+  }
+
+  @Test
+  @Category(NeedsRunner.class)
+  public void getCommitLogFilePatterns() {
+    PCollection<String> patterns =
+        pipeline.apply(
+            "Get CommitLog file patterns",
+            CommitLogTransforms.getCommitLogFilePatterns(commitLogsDir.getAbsolutePath()));
+
+    ImmutableList<String> expectedPatterns =
+        ImmutableList.of(commitLogsDir.getAbsolutePath() + "/commit_diff_until_*");
+
+    PAssert.that(patterns).containsInAnyOrder(expectedPatterns);
+
+    pipeline.run();
+  }
+
+  @Test
+  @Category(NeedsRunner.class)
+  public void getFilesByPatterns() {
+    PCollection<Metadata> fileMetas =
+        pipeline
+            .apply(
+                "File patterns to metadata",
+                Create.of(commitLogsDir.getAbsolutePath() + "/commit_diff_until_*")
+                    .withCoder(StringUtf8Coder.of()))
+            .apply(Transforms.getFilesByPatterns());
+
+    // Transform fileMetas to file names for assertions.
+    PCollection<String> fileNames =
+        fileMetas.apply(
+            "File metadata to path string",
+            ParDo.of(
+                new DoFn<Metadata, String>() {
+                  @ProcessElement
+                  public void processElement(
+                      @Element Metadata metadata, OutputReceiver<String> out) {
+                    out.output(metadata.resourceId().toString());
+                  }
+                }));
+
+    ImmutableList<String> expectedFilenames =
+        ImmutableList.of(firstCommitLogFile.getAbsolutePath());
+
+    PAssert.that(fileNames).containsInAnyOrder(expectedFilenames);
+
+    pipeline.run();
+  }
+
+  @Test
+  @Category(NeedsRunner.class)
+  public void filterCommitLogsByTime() {
+    ImmutableList<String> commitLogFilenames =
+        ImmutableList.of(
+            "/commit_diff_until_2000-01-01T00:00:00.000Z",
+            "/commit_diff_until_2000-01-01T00:00:00.001Z",
+            "/commit_diff_until_2000-01-01T00:00:00.002Z",
+            "/commit_diff_until_2000-01-01T00:00:00.003Z",
+            "/commit_diff_until_2000-01-01T00:00:00.004Z");
+    PCollection<String> filteredFilenames =
+        pipeline
+            .apply(
+                "Generate All Filenames",
+                Create.of(commitLogFilenames).withCoder(StringUtf8Coder.of()))
+            .apply(
+                "Filtered by Time",
+                CommitLogTransforms.filterCommitLogsByTime(
+                    DateTime.parse("2000-01-01T00:00:00.001Z"),
+                    DateTime.parse("2000-01-01T00:00:00.003Z")));
+    PAssert.that(filteredFilenames)
+        .containsInAnyOrder(
+            "/commit_diff_until_2000-01-01T00:00:00.001Z",
+            "/commit_diff_until_2000-01-01T00:00:00.002Z");
+
+    pipeline.run();
+  }
+
+  @Test
+  @Category(NeedsRunner.class)
+  public void loadOneCommitLogFile() {
+    PCollection<VersionedEntity> entities =
+        pipeline
+            .apply(
+                "Get CommitLog file patterns",
+                CommitLogTransforms.getCommitLogFilePatterns(commitLogsDir.getAbsolutePath()))
+            .apply("Find CommitLogs", Transforms.getFilesByPatterns())
+            .apply(CommitLogTransforms.loadCommitLogsFromFiles());
+
+    PCollection<Long> timestamps =
+        entities.apply(
+            "Extract commitTimeMillis",
+            ParDo.of(
+                new DoFn<VersionedEntity, Long>() {
+                  @ProcessElement
+                  public void processElement(
+                      @Element VersionedEntity entity, OutputReceiver<Long> out) {
+                    out.output(entity.commitTimeMills());
+                  }
+                }));
+    PAssert.that(timestamps)
+        .containsInAnyOrder(
+            fakeClock.nowUtc().getMillis() - 2,
+            fakeClock.nowUtc().getMillis() - 1,
+            fakeClock.nowUtc().getMillis() - 1);
+
+    PCollection<Entity> datastoreEntities =
+        entities.apply(
+            "To Datastore Entities",
+            ParDo.of(
+                new DoFn<VersionedEntity, Entity>() {
+                  @ProcessElement
+                  public void processElement(
+                      @Element VersionedEntity entity, OutputReceiver<Entity> out) {
+                    entity.getEntity().ifPresent(out::output);
+                  }
+                }));
+
+    PAssert.that(datastoreEntities).containsInAnyOrder(persistedEntities);
+
+    pipeline.run();
+  }
+}

core/src/test/java/google/registry/beam/initsql/ExportloadingTransformsTest.java

@@ -21,14 +21,15 @@ import static google.registry.testing.DatastoreHelper.newDomainBase;
 import static google.registry.testing.DatastoreHelper.newRegistry;
 
 import com.google.appengine.api.datastore.Entity;
-import com.google.appengine.api.datastore.EntityTranslator;
 import com.google.common.collect.ImmutableList;
-import com.google.storage.onestore.v3.OnestoreEntity.EntityProto;
 import com.googlecode.objectify.Key;
+import google.registry.backup.VersionedEntity;
 import google.registry.model.contact.ContactResource;
 import google.registry.model.domain.DomainBase;
+import google.registry.model.ofy.Ofy;
 import google.registry.model.registry.Registry;
 import google.registry.testing.FakeClock;
+import google.registry.testing.InjectRule;
 import java.io.File;
 import java.io.Serializable;
 import java.util.Collections;
@@ -40,8 +41,8 @@ import org.apache.beam.sdk.testing.TestPipeline;
 import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
+import org.joda.time.DateTime;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
@@ -60,6 +61,8 @@ import org.junit.runners.JUnit4;
 // a wrapper.
 @RunWith(JUnit4.class)
 public class ExportloadingTransformsTest implements Serializable {
+  private static final DateTime START_TIME = DateTime.parse("2000-01-01T00:00:00.0Z");
+
   private static final ImmutableList<Class<?>> ALL_KINDS =
       ImmutableList.of(Registry.class, ContactResource.class, DomainBase.class);
   private static final ImmutableList<String> ALL_KIND_STRS =
@@ -67,6 +70,8 @@ public class ExportloadingTransformsTest implements Serializable {
 
   @Rule public final transient TemporaryFolder exportRootDir = new TemporaryFolder();
 
+  @Rule public final transient InjectRule injectRule = new InjectRule();
+
   @Rule
   public final transient TestPipeline pipeline =
       TestPipeline.create().enableAbandonedNodeEnforcement(true);
@@ -80,8 +85,9 @@ public class ExportloadingTransformsTest implements Serializable {
 
   @Before
   public void beforeEach() throws Exception {
-    fakeClock = new FakeClock();
+    fakeClock = new FakeClock(START_TIME);
     store = new BackupTestStore(fakeClock);
+    injectRule.setStaticField(Ofy.class, "clock", fakeClock);
 
     Registry registry = newRegistry("tld1", "TLD1");
     store.insertOrUpdate(registry);
@@ -107,7 +113,7 @@ public class ExportloadingTransformsTest implements Serializable {
 
   @Test
   @Category(NeedsRunner.class)
-  public void getBackupDataFilePatterns() {
+  public void getExportFilePatterns() {
     PCollection<String> patterns =
         pipeline.apply(
             "Get Datastore file patterns",
@@ -138,7 +144,7 @@ public class ExportloadingTransformsTest implements Serializable {
                         exportDir.getAbsolutePath()
                             + "/all_namespaces/kind_ContactResource/input-*")
                     .withCoder(StringUtf8Coder.of()))
-            .apply(ExportLoadingTransforms.getFilesByPatterns());
+            .apply(Transforms.getFilesByPatterns());
 
     // Transform fileMetas to file names for assertions.
     PCollection<String> fileNames =
@@ -166,25 +172,25 @@ public class ExportloadingTransformsTest implements Serializable {
 
   @Test
   public void loadDataFromFiles() {
-    PCollection<KV<String, byte[]>> taggedRecords =
+    PCollection<VersionedEntity> taggedRecords =
         pipeline
             .apply(
                 "Get Datastore file patterns",
                 ExportLoadingTransforms.getDatastoreExportFilePatterns(
                     exportDir.getAbsolutePath(), ALL_KIND_STRS))
-            .apply("Find Datastore files", ExportLoadingTransforms.getFilesByPatterns())
-            .apply("Load from Datastore files", ExportLoadingTransforms.loadDataFromFiles());
+            .apply("Find Datastore files", Transforms.getFilesByPatterns())
+            .apply("Load from Datastore files", ExportLoadingTransforms.loadExportDataFromFiles());
 
     // Transform bytes to pojo for analysis
     PCollection<Entity> entities =
         taggedRecords.apply(
             "Raw records to Entity",
             ParDo.of(
-                new DoFn<KV<String, byte[]>, Entity>() {
+                new DoFn<VersionedEntity, Entity>() {
                   @ProcessElement
                   public void processElement(
-                      @Element KV<String, byte[]> kv, OutputReceiver<Entity> out) {
-                    out.output(parseBytes(kv.getValue()));
+                      @Element VersionedEntity versionedEntity, OutputReceiver<Entity> out) {
+                    out.output(versionedEntity.getEntity().get());
                   }
                 }));
 
@@ -192,10 +198,4 @@ public class ExportloadingTransformsTest implements Serializable {
 
     pipeline.run();
   }
-
-  private static Entity parseBytes(byte[] record) {
-    EntityProto proto = new EntityProto();
-    proto.parseFrom(record);
-    return EntityTranslator.createFromPb(proto);
-  }
 }

core/src/test/java/google/registry/flows/EppLifecycleDomainTest.java

@@ -127,9 +127,7 @@ public class EppLifecycleDomainTest extends EppTestCase {
             ImmutableMap.of(
                 "DOMAIN", "example.tld",
                 "CRDATE", "2000-06-01T00:02:00Z",
-                // TODO(mcilwain): The exp. date should be restored back to 2002-06-01T00:02:00Z,
-                // but this is old behavior of being 1 year after the moment of the restore.
-                "EXDATE", "2001-07-01T00:03:00Z",
+                "EXDATE", "2002-06-01T00:02:00Z",
                 "UPDATE", "2000-07-01T00:03:00Z"));
 
     assertThatLogoutSucceeds();
@@ -203,11 +201,7 @@ public class EppLifecycleDomainTest extends EppTestCase {
             ImmutableMap.of(
                 "DOMAIN", "example.tld",
                 "CRDATE", "2000-06-01T00:02:00Z",
-                // TODO(mcilwain): The exp. date should be 2003-06-01T00:02:00Z, the same as its
-                // value prior to the deletion, because the year that was taken off when the
-                // autorenew was canceled will be re-added in renewal during the restore.
-                // For now though, the current behavior is 1 year after restore.
-                "EXDATE", "2003-07-05T00:03:00Z",
+                "EXDATE", "2003-06-01T00:02:00Z",
                 "UPDATE", "2002-07-05T00:03:00Z"));
 
     assertThatLogoutSucceeds();
@@ -289,10 +283,7 @@ public class EppLifecycleDomainTest extends EppTestCase {
             ImmutableMap.of(
                 "DOMAIN", "example.tld",
                 "CRDATE", "2000-06-01T00:02:00Z",
-                // TODO(mcilwain): The exp. date should be 2002-06-01T00:02:00Z, which is the
-                // current registration expiration time on the (deleted) domain, but for now is
-                // 1 year after restore.
-                "EXDATE", "2001-06-20T00:00:00Z",
+                "EXDATE", "2002-06-01T00:02:00Z",
                 "UPDATE", "2000-06-20T00:00:00Z"));
 
     assertThatLogoutSucceeds();

core/src/test/java/google/registry/flows/custom/TestDomainPricingCustomLogic.java

@@ -40,7 +40,7 @@ public class TestDomainPricingCustomLogic extends DomainPricingCustomLogic {
   public FeesAndCredits customizeRenewPrice(RenewPriceParameters priceParameters) {
     return priceParameters.domainName().toString().startsWith("costly-renew")
         ? addCustomFee(
-            priceParameters.feesAndCredits(), Fee.create(ONE_HUNDRED_BUCKS, FeeType.RENEW))
+            priceParameters.feesAndCredits(), Fee.create(ONE_HUNDRED_BUCKS, FeeType.RENEW, true))
         : priceParameters.feesAndCredits();
   }
 
@@ -48,7 +48,7 @@ public class TestDomainPricingCustomLogic extends DomainPricingCustomLogic {
   public FeesAndCredits customizeTransferPrice(TransferPriceParameters priceParameters) {
     return priceParameters.domainName().toString().startsWith("expensive")
         ? addCustomFee(
-            priceParameters.feesAndCredits(), Fee.create(ONE_HUNDRED_BUCKS, FeeType.TRANSFER))
+            priceParameters.feesAndCredits(), Fee.create(ONE_HUNDRED_BUCKS, FeeType.TRANSFER, true))
         : priceParameters.feesAndCredits();
   }
 
@@ -56,7 +56,7 @@ public class TestDomainPricingCustomLogic extends DomainPricingCustomLogic {
   public FeesAndCredits customizeUpdatePrice(UpdatePriceParameters priceParameters) {
     return priceParameters.domainName().toString().startsWith("non-free-update")
         ? addCustomFee(
-            priceParameters.feesAndCredits(), Fee.create(ONE_HUNDRED_BUCKS, FeeType.UPDATE))
+            priceParameters.feesAndCredits(), Fee.create(ONE_HUNDRED_BUCKS, FeeType.UPDATE, true))
         : priceParameters.feesAndCredits();
   }
 

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

@@ -72,6 +72,7 @@ import google.registry.model.reporting.DomainTransactionRecord.TransactionReport
 import google.registry.model.reporting.HistoryEntry;
 import java.util.Map;
 import org.joda.money.Money;
+import org.joda.time.DateTime;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -93,6 +94,10 @@ public class DomainRestoreRequestFlowTest
   }
 
   void persistPendingDeleteDomain() throws Exception {
+    persistPendingDeleteDomain(clock.nowUtc().plusYears(5).plusDays(45));
+  }
+
+  void persistPendingDeleteDomain(DateTime expirationTime) throws Exception {
     DomainBase domain = newDomainBase(getUniqueIdFromCommand());
     HistoryEntry historyEntry =
         persistResource(
@@ -103,7 +108,7 @@ public class DomainRestoreRequestFlowTest
     persistResource(
         domain
             .asBuilder()
-            .setRegistrationExpirationTime(clock.nowUtc().plusYears(5).plusDays(45))
+            .setRegistrationExpirationTime(expirationTime)
             .setDeletionTime(clock.nowUtc().plusDays(35))
             .addGracePeriod(
                 GracePeriod.create(
@@ -129,9 +134,10 @@ public class DomainRestoreRequestFlowTest
   }
 
   @Test
-  public void testSuccess() throws Exception {
+  public void testSuccess_expiryStillInFuture_notExtended() throws Exception {
     setEppInput("domain_update_restore_request.xml", ImmutableMap.of("DOMAIN", "example.tld"));
-    persistPendingDeleteDomain();
+    DateTime expirationTime = clock.nowUtc().plusYears(5).plusDays(45);
+    persistPendingDeleteDomain(expirationTime);
     assertTransactionalFlow(true);
     // Double check that we see a poll message in the future for when the delete happens.
     assertThat(getPollMessages("TheRegistrar", clock.nowUtc().plusMonths(1))).hasSize(1);
@@ -140,11 +146,79 @@ public class DomainRestoreRequestFlowTest
     HistoryEntry historyEntryDomainRestore =
         getOnlyHistoryEntryOfType(domain, HistoryEntry.Type.DOMAIN_RESTORE);
     assertThat(ofy().load().key(domain.getAutorenewBillingEvent()).now().getEventTime())
-        .isEqualTo(clock.nowUtc().plusYears(1));
+        .isEqualTo(expirationTime);
+    assertAboutDomains()
+        .that(domain)
+        // New expiration time should be the same as from before the deletion.
+        .hasRegistrationExpirationTime(expirationTime)
+        .and()
+        .doesNotHaveStatusValue(StatusValue.PENDING_DELETE)
+        .and()
+        .hasDeletionTime(END_OF_TIME)
+        .and()
+        .hasOneHistoryEntryEachOfTypes(
+            HistoryEntry.Type.DOMAIN_DELETE, HistoryEntry.Type.DOMAIN_RESTORE)
+        .and()
+        .hasLastEppUpdateTime(clock.nowUtc())
+        .and()
+        .hasLastEppUpdateClientId("TheRegistrar");
+    assertThat(domain.getGracePeriods()).isEmpty();
+    assertDnsTasksEnqueued("example.tld");
+    // The poll message for the delete should now be gone. The only poll message should be the new
+    // autorenew poll message.
+    assertPollMessages(
+        "TheRegistrar",
+        new PollMessage.Autorenew.Builder()
+            .setTargetId("example.tld")
+            .setClientId("TheRegistrar")
+            .setEventTime(domain.getRegistrationExpirationTime())
+            .setAutorenewEndTime(END_OF_TIME)
+            .setMsg("Domain was auto-renewed.")
+            .setParent(historyEntryDomainRestore)
+            .build());
+    // There should be a onetime for the restore and a new recurring billing event, but no renew
+    // onetime.
+    assertBillingEvents(
+        new BillingEvent.Recurring.Builder()
+            .setReason(Reason.RENEW)
+            .setFlags(ImmutableSet.of(Flag.AUTO_RENEW))
+            .setTargetId("example.tld")
+            .setClientId("TheRegistrar")
+            .setEventTime(expirationTime)
+            .setRecurrenceEndTime(END_OF_TIME)
+            .setParent(historyEntryDomainRestore)
+            .build(),
+        new BillingEvent.OneTime.Builder()
+            .setReason(Reason.RESTORE)
+            .setTargetId("example.tld")
+            .setClientId("TheRegistrar")
+            .setCost(Money.of(USD, 17))
+            .setPeriodYears(1)
+            .setEventTime(clock.nowUtc())
+            .setBillingTime(clock.nowUtc())
+            .setParent(historyEntryDomainRestore)
+            .build());
+  }
+
+  @Test
+  public void testSuccess_expiryInPast_extendedByOneYear() throws Exception {
+    setEppInput("domain_update_restore_request.xml", ImmutableMap.of("DOMAIN", "example.tld"));
+    DateTime expirationTime = clock.nowUtc().minusDays(20);
+    DateTime newExpirationTime = expirationTime.plusYears(1);
+    persistPendingDeleteDomain(expirationTime);
+    assertTransactionalFlow(true);
+    // Double check that we see a poll message in the future for when the delete happens.
+    assertThat(getPollMessages("TheRegistrar", clock.nowUtc().plusMonths(1))).hasSize(1);
+    runFlowAssertResponse(loadFile("generic_success_response.xml"));
+    DomainBase domain = reloadResourceByForeignKey();
+    HistoryEntry historyEntryDomainRestore =
+        getOnlyHistoryEntryOfType(domain, HistoryEntry.Type.DOMAIN_RESTORE);
+    assertThat(ofy().load().key(domain.getAutorenewBillingEvent()).now().getEventTime())
+        .isEqualTo(newExpirationTime);
     assertAboutDomains()
         .that(domain)
         // New expiration time should be exactly a year from now.
-        .hasRegistrationExpirationTime(clock.nowUtc().plusYears(1))
+        .hasRegistrationExpirationTime(newExpirationTime)
         .and()
         .doesNotHaveStatusValue(StatusValue.PENDING_DELETE)
         .and()
@@ -178,7 +252,7 @@ public class DomainRestoreRequestFlowTest
             .setFlags(ImmutableSet.of(Flag.AUTO_RENEW))
             .setTargetId("example.tld")
             .setClientId("TheRegistrar")
-            .setEventTime(domain.getRegistrationExpirationTime())
+            .setEventTime(newExpirationTime)
             .setRecurrenceEndTime(END_OF_TIME)
             .setParent(historyEntryDomainRestore)
             .build(),

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

@@ -108,15 +108,16 @@ public class DomainTransferFlowTestCase<F extends Flow, R extends EppResource>
             clock.nowUtc(),
             DateTime.parse("1999-04-03T22:00:00.0Z"),
             REGISTRATION_EXPIRATION_TIME);
-    subordinateHost = persistResource(
-        new HostResource.Builder()
-            .setRepoId("2-".concat(Ascii.toUpperCase(tld)))
-            .setFullyQualifiedHostName("ns1." + label + "." + tld)
-            .setPersistedCurrentSponsorClientId("TheRegistrar")
-            .setCreationClientId("TheRegistrar")
-            .setCreationTimeForTest(DateTime.parse("1999-04-03T22:00:00.0Z"))
-            .setSuperordinateDomain(domain.createVKey())
-            .build());
+    subordinateHost =
+        persistResource(
+            new HostResource.Builder()
+                .setRepoId("2-".concat(Ascii.toUpperCase(tld)))
+                .setFullyQualifiedHostName("ns1." + label + "." + tld)
+                .setPersistedCurrentSponsorClientId("TheRegistrar")
+                .setCreationClientId("TheRegistrar")
+                .setCreationTimeForTest(DateTime.parse("1999-04-03T22:00:00.0Z"))
+                .setSuperordinateDomain(domain.createVKey())
+                .build());
     domain =
         persistResource(
             domain

core/src/test/java/google/registry/persistence/transaction/TransactionManagerTest.java

@@ -101,8 +101,7 @@ public class TransactionManagerTest {
     assertThrows(
         RuntimeException.class,
         () ->
-            tm()
-                .transact(
+            tm().transact(
                     () -> {
                       tm().saveNew(theEntity);
                       throw new RuntimeException();

core/src/test/java/google/registry/ui/server/registrar/RegistryLockPostActionTest.java

@@ -28,6 +28,7 @@ import static google.registry.ui.server.registrar.RegistryLockGetActionTest.user
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
 
 import com.google.appengine.api.users.User;
 import com.google.common.collect.ImmutableList;
@@ -56,6 +57,7 @@ import java.util.Map;
 import java.util.Optional;
 import java.util.UUID;
 import javax.mail.internet.InternetAddress;
+import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import org.joda.time.Duration;
 import org.junit.Before;
@@ -74,7 +76,7 @@ public final class RegistryLockPostActionTest {
   private static final String EMAIL_MESSAGE_TEMPLATE =
       "Please click the link below to perform the lock \\/ unlock action on domain example.tld. "
           + "Note: this code will expire in one hour.\n\n"
-          + "https:\\/\\/localhost\\/registry-lock-verify\\?lockVerificationCode="
+          + "https:\\/\\/registrarconsole.tld\\/registry-lock-verify\\?lockVerificationCode="
           + "[0-9a-zA-Z_\\-]+&isLock=(true|false)";
 
   private final FakeClock clock = new FakeClock();
@@ -93,6 +95,7 @@ public final class RegistryLockPostActionTest {
   private RegistryLockPostAction action;
 
   @Mock SendEmailService emailService;
+  @Mock HttpServletRequest mockRequest;
   @Mock HttpServletResponse mockResponse;
 
   @Before
@@ -103,6 +106,8 @@ public final class RegistryLockPostActionTest {
     domain = persistResource(newDomainBase("example.tld"));
     outgoingAddress = new InternetAddress("domain-registry@example.com");
 
+    when(mockRequest.getServerName()).thenReturn("registrarconsole.tld");
+
     action =
         createAction(
             AuthResult.create(AuthLevel.USER, UserAuthInfo.create(userWithLockPermission, false)));
@@ -432,6 +437,7 @@ public final class RegistryLockPostActionTest {
             AsyncTaskEnqueuerTest.createForTesting(
                 mock(AppEngineServiceUtils.class), clock, Duration.ZERO));
     return new RegistryLockPostAction(
+        mockRequest,
         jsonActionRunner,
         authResult,
         registrarAccessor,

core/src/test/resources/google/registry/flows/domain_info_response_fakesite_restored_ok.xml

@@ -21,7 +21,7 @@
         <domain:crDate>2000-06-01T00:04:00Z</domain:crDate>
         <domain:upID>NewRegistrar</domain:upID>
         <domain:upDate>2002-05-30T01:03:00Z</domain:upDate>
-        <domain:exDate>2003-05-30T01:03:00Z</domain:exDate>
+        <domain:exDate>2002-06-01T00:04:00Z</domain:exDate>
         <domain:authInfo>
           <domain:pw>2fooBAR</domain:pw>
         </domain:authInfo>

networking/src/main/java/google/registry/networking/handler/SslServerInitializer.java

@@ -25,6 +25,7 @@ import io.netty.channel.ChannelHandler.Sharable;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.embedded.EmbeddedChannel;
 import io.netty.handler.ssl.ClientAuth;
+import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.ssl.SslHandler;
 import io.netty.handler.ssl.SslProvider;
@@ -33,9 +34,11 @@ import io.netty.util.AttributeKey;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.Promise;
 import java.security.PrivateKey;
+import java.security.PublicKey;
 import java.security.cert.CertificateExpiredException;
 import java.security.cert.CertificateNotYetValidException;
 import java.security.cert.X509Certificate;
+import java.security.interfaces.RSAPublicKey;
 import java.util.function.Supplier;
 import javax.net.ssl.SSLSession;
 
@@ -72,6 +75,7 @@ public class SslServerInitializer<C extends Channel> extends ChannelInitializer<
   // change when the artifacts on GCS changes.
   private final Supplier<PrivateKey> privateKeySupplier;
   private final Supplier<ImmutableList<X509Certificate>> certificatesSupplier;
+  private final ImmutableList<String> supportedSslVersions;
 
   public SslServerInitializer(
       boolean requireClientCert,
@@ -88,19 +92,27 @@ public class SslServerInitializer<C extends Channel> extends ChannelInitializer<
     this.sslProvider = sslProvider;
     this.privateKeySupplier = privateKeySupplier;
     this.certificatesSupplier = certificatesSupplier;
+    this.supportedSslVersions =
+        sslProvider == SslProvider.OPENSSL
+            ? ImmutableList.of("TLSv1.3", "TLSv1.2", "TLSv1.1", "TLSv1")
+            // JDK support for TLS 1.3 won't be available until 2020-07-14 at the earliest.
+            // See: https://java.com/en/jre-jdk-cryptoroadmap.html
+            : ImmutableList.of("TLSv1.2", "TLSv1.1", "TLSv1");
   }
 
   @Override
   protected void initChannel(C channel) throws Exception {
-    SslHandler sslHandler =
+    SslContext sslContext =
         SslContextBuilder.forServer(
                 privateKeySupplier.get(),
                 certificatesSupplier.get().toArray(new X509Certificate[0]))
             .sslProvider(sslProvider)
             .trustManager(InsecureTrustManagerFactory.INSTANCE)
             .clientAuth(requireClientCert ? ClientAuth.REQUIRE : ClientAuth.NONE)
-            .build()
-            .newHandler(channel.alloc());
+            .protocols(supportedSslVersions)
+            .build();
+    logger.atInfo().log("Available Cipher Suites: %s", sslContext.cipherSuites());
+    SslHandler sslHandler = sslContext.newHandler(channel.alloc());
     if (requireClientCert) {
       Promise<X509Certificate> clientCertificatePromise = channel.eventLoop().newPromise();
       Future<Channel> unusedFuture =
@@ -112,18 +124,29 @@ public class SslServerInitializer<C extends Channel> extends ChannelInitializer<
                       SSLSession sslSession = sslHandler.engine().getSession();
                       X509Certificate clientCertificate =
                           (X509Certificate) sslSession.getPeerCertificates()[0];
+                      PublicKey clientPublicKey = clientCertificate.getPublicKey();
+                      // Note that for non-RSA keys the length would be -1.
+                      int clientCertificateLength = -1;
+                      if (clientPublicKey instanceof RSAPublicKey) {
+                        clientCertificateLength =
+                            ((RSAPublicKey) clientPublicKey).getModulus().bitLength();
+                      }
                       logger.atInfo().log(
                           "--SSL Information--\n"
                               + "Client Certificate Hash: %s\n"
                               + "SSL Protocol: %s\n"
                               + "Cipher Suite: %s\n"
                               + "Not Before: %s\n"
-                              + "Not After: %s\n",
+                              + "Not After: %s\n"
+                              + "Client Certificate Type: %s\n"
+                              + "Client Certificate Length: %s\n",
                           getCertificateHash(clientCertificate),
                           sslSession.getProtocol(),
                           sslSession.getCipherSuite(),
                           clientCertificate.getNotBefore(),
-                          clientCertificate.getNotAfter());
+                          clientCertificate.getNotAfter(),
+                          clientPublicKey.getClass().getName(),
+                          clientCertificateLength);
                       try {
                         clientCertificate.checkValidity();
                       } catch (CertificateNotYetValidException | CertificateExpiredException e) {