Fail genenerate invoices job when billing events not finished expanding (#1791)

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

@@ -267,7 +267,7 @@
     about 2 hours to complete, so we give 11 hours to be safe. Normally, we give 24+ hours (see
     icannReportingStaging), but the invoicing team prefers receiving the e-mail on the first of
     each month. -->
-    <schedule>1 of month 17:00</schedule>
+    <schedule>1 of month 19:00</schedule>
     <target>backend</target>
   </cron>
 

core/src/main/java/google/registry/reporting/billing/GenerateInvoicesAction.java

@@ -15,7 +15,10 @@
 package google.registry.reporting.billing;
 
 import static google.registry.beam.BeamUtils.createJobName;
+import static google.registry.model.common.Cursor.CursorType.RECURRING_BILLING;
+import static google.registry.persistence.transaction.TransactionManagerFactory.tm;
 import static google.registry.request.Action.Method.POST;
+import static google.registry.util.DateTimeUtils.START_OF_TIME;
 import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
 import static javax.servlet.http.HttpServletResponse.SC_OK;
 
@@ -29,6 +32,7 @@ import com.google.common.flogger.FluentLogger;
 import com.google.common.net.MediaType;
 import google.registry.config.RegistryConfig.Config;
 import google.registry.config.RegistryEnvironment;
+import google.registry.model.common.Cursor;
 import google.registry.persistence.PersistenceModule;
 import google.registry.reporting.ReportingModule;
 import google.registry.request.Action;
@@ -40,6 +44,7 @@ import google.registry.util.Clock;
 import google.registry.util.CloudTasksUtils;
 import java.io.IOException;
 import javax.inject.Inject;
+import org.joda.time.DateTime;
 import org.joda.time.Duration;
 import org.joda.time.YearMonth;
 
@@ -108,6 +113,19 @@ public class GenerateInvoicesAction implements Runnable {
     response.setContentType(MediaType.PLAIN_TEXT_UTF_8);
     logger.atInfo().log("Launching invoicing pipeline for %s.", yearMonth);
     try {
+      DateTime currentCursorTime =
+          tm().transact(
+                  () ->
+                      tm().loadByKeyIfPresent(Cursor.createGlobalVKey(RECURRING_BILLING))
+                          .orElse(Cursor.createGlobal(RECURRING_BILLING, START_OF_TIME))
+                          .getCursorTime());
+
+      if (yearMonth.getMonthOfYear() >= currentCursorTime.getMonthOfYear()) {
+        throw new IllegalStateException(
+            "Latest billing events expansion cycle hasn't finished yet, terminating invoicing"
+                + " pipeline");
+      }
+
       LaunchFlexTemplateParameter parameter =
           new LaunchFlexTemplateParameter()
               .setJobName(createJobName("invoicing", clock))
@@ -150,7 +168,7 @@ public class GenerateInvoicesAction implements Runnable {
       }
       response.setStatus(SC_OK);
       response.setPayload(String.format("Launched invoicing pipeline: %s", jobId));
-    } catch (IOException e) {
+    } catch (IOException | IllegalStateException e) {
       logger.atWarning().withCause(e).log("Template Launch failed.");
       emailUtils.sendAlertEmail(String.format("Pipeline Launch failed due to %s", e.getMessage()));
       response.setStatus(SC_INTERNAL_SERVER_ERROR);