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);

core/src/test/java/google/registry/reporting/billing/GenerateInvoicesActionTest.java

@@ -15,6 +15,8 @@
 package google.registry.reporting.billing;
 
 import static com.google.common.truth.Truth.assertThat;
+import static google.registry.model.common.Cursor.CursorType.RECURRING_BILLING;
+import static google.registry.testing.DatabaseHelper.persistResource;
 import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
 import static javax.servlet.http.HttpServletResponse.SC_OK;
 import static org.mockito.Mockito.mock;
@@ -24,6 +26,7 @@ import static org.mockito.Mockito.when;
 import com.google.cloud.tasks.v2.HttpMethod;
 import com.google.common.net.MediaType;
 import google.registry.beam.BeamActionTestBase;
+import google.registry.model.common.Cursor;
 import google.registry.reporting.ReportingModule;
 import google.registry.testing.AppEngineExtension;
 import google.registry.testing.CloudTasksHelper;
@@ -31,8 +34,10 @@ import google.registry.testing.CloudTasksHelper.TaskMatcher;
 import google.registry.testing.FakeClock;
 import google.registry.util.CloudTasksUtils;
 import java.io.IOException;
+import org.joda.time.DateTime;
 import org.joda.time.Duration;
 import org.joda.time.YearMonth;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
@@ -49,6 +54,13 @@ class GenerateInvoicesActionTest extends BeamActionTestBase {
   private CloudTasksUtils cloudTasksUtils = cloudTasksHelper.getTestCloudTasksUtils();
   private GenerateInvoicesAction action;
 
+  @BeforeEach
+  @Override
+  protected void beforeEach() throws Exception {
+    super.beforeEach();
+    persistResource(Cursor.createGlobal(RECURRING_BILLING, DateTime.parse("2017-11-30TZ")));
+  }
+
   @Test
   void testLaunchTemplateJob_withPublish() throws Exception {
     action =
@@ -129,4 +141,31 @@ class GenerateInvoicesActionTest extends BeamActionTestBase {
     verify(emailUtils).sendAlertEmail("Pipeline Launch failed due to Pipeline error");
     cloudTasksHelper.assertNoTasksEnqueued("beam-reporting");
   }
+
+  @Test
+  void testFailsToGenerateInvoicesNotExpandedBillingEvents() throws Exception {
+    persistResource(Cursor.createGlobal(RECURRING_BILLING, DateTime.parse("2017-10-30TZ")));
+    action =
+        new GenerateInvoicesAction(
+            "test-project",
+            "test-region",
+            "staging_bucket",
+            "billing_bucket",
+            "REG-INV",
+            false,
+            new YearMonth(2017, 10),
+            emailUtils,
+            cloudTasksUtils,
+            clock,
+            response,
+            dataflow);
+    action.run();
+    assertThat(response.getContentType()).isEqualTo(MediaType.PLAIN_TEXT_UTF_8);
+    assertThat(response.getStatus()).isEqualTo(SC_INTERNAL_SERVER_ERROR);
+    assertThat(response.getPayload())
+        .isEqualTo(
+            "Pipeline launch failed: Latest billing events expansion cycle hasn't finished yet,"
+                + " terminating invoicing pipeline");
+    cloudTasksHelper.assertNoTasksEnqueued("beam-reporting");
+  }
 }