From b1e42cfd5ee0bd46f4ed0f06d752d737715f50b0 Mon Sep 17 00:00:00 2001 From: gbrodman Date: Tue, 7 Jul 2026 14:05:48 -0400 Subject: [PATCH] Verify billing recurrence expansion is caught up before invoicing (#3121) --- .../billing/GenerateInvoicesAction.java | 24 +++++++ .../billing/GenerateInvoicesActionTest.java | 66 +++++++++++++++++++ 2 files changed, 90 insertions(+) diff --git a/core/src/main/java/google/registry/reporting/billing/GenerateInvoicesAction.java b/core/src/main/java/google/registry/reporting/billing/GenerateInvoicesAction.java index bb5f44de5..4fd8acce1 100644 --- a/core/src/main/java/google/registry/reporting/billing/GenerateInvoicesAction.java +++ b/core/src/main/java/google/registry/reporting/billing/GenerateInvoicesAction.java @@ -14,10 +14,14 @@ package google.registry.reporting.billing; +import static com.google.common.base.Preconditions.checkState; 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 jakarta.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR; import static jakarta.servlet.http.HttpServletResponse.SC_OK; +import static java.time.ZoneOffset.UTC; import com.google.api.services.dataflow.Dataflow; import com.google.api.services.dataflow.model.LaunchFlexTemplateParameter; @@ -29,6 +33,7 @@ import com.google.common.flogger.FluentLogger; import com.google.common.net.MediaType; import google.registry.batch.CloudTasksUtils; import google.registry.config.RegistryConfig.Config; +import google.registry.model.common.Cursor; import google.registry.persistence.PersistenceModule; import google.registry.reporting.ReportingModule; import google.registry.request.Action; @@ -40,7 +45,9 @@ import google.registry.util.RegistryEnvironment; import jakarta.inject.Inject; import java.io.IOException; import java.time.Duration; +import java.time.Instant; import java.time.YearMonth; +import java.util.Optional; /** * Invokes the {@code InvoicingPipeline} beam template via the REST api, and enqueues the {@link @@ -107,6 +114,7 @@ public class GenerateInvoicesAction implements Runnable { response.setContentType(MediaType.PLAIN_TEXT_UTF_8); logger.atInfo().log("Launching invoicing pipeline for %s.", yearMonth); try { + checkBillingRecurrenceCursor(); LaunchFlexTemplateParameter parameter = new LaunchFlexTemplateParameter() .setJobName(createJobName("invoicing", clock)) @@ -156,4 +164,20 @@ public class GenerateInvoicesAction implements Runnable { response.setPayload(String.format("Pipeline launch failed: %s", e.getMessage())); } } + + private void checkBillingRecurrenceCursor() { + Optional previousCursor = + tm().transact(() -> tm().loadByKeyIfPresent(Cursor.createGlobalVKey(RECURRING_BILLING))); + checkState( + previousCursor.isPresent(), + "BillingRecurrence expansion cursor is not present. Run ExpandBillingRecurrencesAction."); + Instant startOfNextMonth = yearMonth.plusMonths(1).atDay(1).atStartOfDay(UTC).toInstant(); + Instant previousCursorTime = previousCursor.get().getCursorTime(); + checkState( + !previousCursorTime.isBefore(startOfNextMonth), + "BillingRecurrence expansion cursor (%s) is before the start of the next month (%s). " + + "Run ExpandBillingRecurrencesAction.", + previousCursorTime, + startOfNextMonth); + } } diff --git a/core/src/test/java/google/registry/reporting/billing/GenerateInvoicesActionTest.java b/core/src/test/java/google/registry/reporting/billing/GenerateInvoicesActionTest.java index a77cc7d61..4b96effd1 100644 --- a/core/src/test/java/google/registry/reporting/billing/GenerateInvoicesActionTest.java +++ b/core/src/test/java/google/registry/reporting/billing/GenerateInvoicesActionTest.java @@ -15,8 +15,11 @@ 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 jakarta.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR; import static jakarta.servlet.http.HttpServletResponse.SC_OK; +import static org.mockito.ArgumentMatchers.startsWith; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -25,6 +28,7 @@ import com.google.cloud.tasks.v2.HttpMethod; import com.google.common.net.MediaType; import google.registry.batch.CloudTasksUtils; import google.registry.beam.BeamActionTestBase; +import google.registry.model.common.Cursor; import google.registry.persistence.transaction.JpaTestExtensions; import google.registry.persistence.transaction.JpaTestExtensions.JpaIntegrationTestExtension; import google.registry.reporting.ReportingModule; @@ -33,6 +37,7 @@ import google.registry.testing.CloudTasksHelper.TaskMatcher; import google.registry.testing.FakeClock; import java.io.IOException; import java.time.Duration; +import java.time.Instant; import java.time.YearMonth; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; @@ -50,8 +55,13 @@ class GenerateInvoicesActionTest extends BeamActionTestBase { private CloudTasksUtils cloudTasksUtils = cloudTasksHelper.getTestCloudTasksUtils(); private GenerateInvoicesAction action; + private void setCursor(Instant cursorTime) { + persistResource(Cursor.createGlobal(RECURRING_BILLING, cursorTime)); + } + @Test void testLaunchTemplateJob_withPublish() throws Exception { + setCursor(Instant.parse("2017-11-01T00:00:00Z")); action = new GenerateInvoicesAction( "test-project", @@ -84,6 +94,7 @@ class GenerateInvoicesActionTest extends BeamActionTestBase { @Test void testLaunchTemplateJob_withoutPublish() throws Exception { + setCursor(Instant.parse("2017-11-01T00:00:00Z")); action = new GenerateInvoicesAction( "test-project", @@ -107,6 +118,7 @@ class GenerateInvoicesActionTest extends BeamActionTestBase { @Test void testCaughtIOException() throws IOException { + setCursor(Instant.parse("2017-11-01T00:00:00Z")); when(launch.execute()).thenThrow(new IOException("Pipeline error")); action = new GenerateInvoicesAction( @@ -128,4 +140,58 @@ class GenerateInvoicesActionTest extends BeamActionTestBase { verify(emailUtils).sendAlertEmail("Pipeline Launch failed due to Pipeline error"); cloudTasksHelper.assertNoTasksEnqueued("beam-reporting"); } + + @Test + void testFailure_cursorLagging() { + setCursor(Instant.parse("2017-10-31T23:59:59.999Z")); + action = + new GenerateInvoicesAction( + "test-project", + "test-region", + "staging_bucket", + "billing_bucket", + "REG-INV", + false, + YearMonth.of(2017, 10), + emailUtils, + cloudTasksUtils, + clock, + response, + dataflow); + action.run(); + assertThat(response.getStatus()).isEqualTo(SC_INTERNAL_SERVER_ERROR); + assertThat(response.getPayload()).contains("Pipeline launch failed"); + assertThat(response.getPayload()).contains("BillingRecurrence expansion cursor"); + verify(emailUtils) + .sendAlertEmail( + startsWith("Pipeline Launch failed due to BillingRecurrence expansion cursor")); + cloudTasksHelper.assertNoTasksEnqueued("beam-reporting"); + } + + @Test + void testFailure_cursorMissing() { + // Do not set cursor, should default to START_INSTANT (1970) + action = + new GenerateInvoicesAction( + "test-project", + "test-region", + "staging_bucket", + "billing_bucket", + "REG-INV", + false, + YearMonth.of(2017, 10), + emailUtils, + cloudTasksUtils, + clock, + response, + dataflow); + action.run(); + assertThat(response.getStatus()).isEqualTo(SC_INTERNAL_SERVER_ERROR); + assertThat(response.getPayload()).contains("Pipeline launch failed"); + assertThat(response.getPayload()).contains("BillingRecurrence expansion cursor"); + verify(emailUtils) + .sendAlertEmail( + startsWith("Pipeline Launch failed due to BillingRecurrence expansion cursor")); + cloudTasksHelper.assertNoTasksEnqueued("beam-reporting"); + } }