[APP-974] scheduled sp locks

README.md

@@ -143,6 +143,22 @@ This setting does not enable diagnostic `@debug` result sets. See
 [`documentation/EventProcedureJobFlowLogging.md`](documentation/EventProcedureJobFlowLogging.md)
 for operational guidance and stored-procedure coverage behavior.
 
+### Scheduled cleanup coordination
+
+The scheduled event-metric and LAB100 cleanup procedures use separate SQL
+Server application locks to prevent duplicate execution by multiple service
+pods. A pod that cannot acquire its procedure's exclusive session lock returns
+immediately and logs `Skipped <procedure> because it's already running`; the
+next cron occurrence retries it. A successful execution logs `Stored proc
+execution completed: <procedure>`. A `-1` procedure result raises a scheduler
+error rather than a completion log.
+
+The lock resources are `RTR:scheduled:event-metric-cleanup` and
+`RTR:scheduled:lab100-cleanup`. They use zero timeout, so contention does not
+block the scheduler, and the resources are independent, so the two cleanup
+jobs can run concurrently. This is duplicate-execution coordination only, not
+a distributed work queue or post-processing-cache coordinator.
+
 ### Windows Note (Testcontainers EOF Error)
 
 If you see `unexpected EOF` while running containerized tests on Windows, this is usually caused by Testcontainers trying to copy the entire project directory into a container and hitting a locked file.

reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/actrelationship/ActRelationshipProcessor.java

@@ -49,7 +49,7 @@ public void process(String message, long batchId) {
       // extract uid and relationship type from message
       sourceActUid = getSourceActUid(message, operation);
       String typeCd = getTypeCd(message, operation);
-      
+
       // call the relevant handler based on the relationship type
       if (isVaccinationRelationship(typeCd)) {
         investigationService.processVaccination(message, false, sourceActUid);

reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/repository/PostProcRepository.java

@@ -127,9 +127,14 @@ void executeStoredProcForBackfill(
   @Query(value = "EXEC sp_nrt_backfill_event :statusCd", nativeQuery = true)
   List<BackfillData> executeBackfillEvent(@Param("statusCd") String statusCd);
 
-  @Procedure("sp_event_metric_cleanup_postprocessing")
-  void executeEventMetricCleanup();
+  @Query(
+      value =
+          "DECLARE @status int; EXEC @status = dbo.sp_event_metric_cleanup_postprocessing; SELECT @status",
+      nativeQuery = true)
+  Integer executeEventMetricCleanup();
 
-  @Procedure("sp_lab100_cleanup")
-  void executeLab100Cleanup();
+  @Query(
+      value = "DECLARE @status int; EXEC @status = dbo.sp_lab100_cleanup; SELECT @status",
+      nativeQuery = true)
+  Integer executeLab100Cleanup();
 }

reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/PostProcessingService.java

@@ -76,6 +76,7 @@
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import lombok.NonNull;
@@ -1569,15 +1570,22 @@ private void completeLog(String sp) {
 
   @Scheduled(cron = "${service.schedule.event-metric-cleanup}")
   protected void eventMetricCleanup() {
-    logger.info("Running event metric cleanup...");
-    postProcRepository.executeEventMetricCleanup();
-    logger.info(SP_EXECUTION_COMPLETED, "sp_event_metric_cleanup_postprocessing");
+    processScheduledProcedure(
+        "sp_event_metric_cleanup_postprocessing", postProcRepository::executeEventMetricCleanup);
   }
 
   @Scheduled(cron = "${service.schedule.lab100-cleanup}")
   protected void lab100Cleanup() {
-    logger.info("Running lab100 cleanup...");
-    postProcRepository.executeLab100Cleanup();
-    logger.info(SP_EXECUTION_COMPLETED, "sp_lab100_cleanup");
+    processScheduledProcedure("sp_lab100_cleanup", postProcRepository::executeLab100Cleanup);
+  }
+
+  private void processScheduledProcedure(String name, Supplier<Integer> scheduledProcedure) {
+    logger.info("Running {}...", name);
+    switch (ScheduledExecutionStatus.fromReturnCode(scheduledProcedure.get())) {
+      case COMPLETED -> logger.info(SP_EXECUTION_COMPLETED, name);
+      case SKIPPED -> logger.info("Skipped {} because it's already running", name);
+      case FAILED -> throw new DataProcessingException(name + " reported a cleanup failure");
+      default -> throw new DataProcessingException(name + " encountered an unknown status");
+    }
   }
 }

reporting-pipeline-service/src/main/java/gov/cdc/nbs/report/pipeline/postprocessing/service/ScheduledExecutionStatus.java

@@ -0,0 +1,27 @@
+package gov.cdc.nbs.report.pipeline.postprocessing.service;
+
+import gov.cdc.nbs.report.pipeline.util.DataProcessingException;
+
+enum ScheduledExecutionStatus {
+  COMPLETED(1),
+  SKIPPED(-2),
+  FAILED(-1);
+
+  private final int returnCode;
+
+  ScheduledExecutionStatus(int returnCode) {
+    this.returnCode = returnCode;
+  }
+
+  static ScheduledExecutionStatus fromReturnCode(Integer returnCode) {
+    if (returnCode == null) {
+      throw new DataProcessingException("Cleanup procedure did not return an execution status");
+    }
+    for (ScheduledExecutionStatus status : values()) {
+      if (status.returnCode == returnCode) {
+        return status;
+      }
+    }
+    throw new DataProcessingException("Unexpected cleanup procedure return code: " + returnCode);
+  }
+}

reporting-pipeline-service/src/main/resources/db/changelog/migrations/v7.13/rdb/routines/345-sp_event_metric_cleanup_postprocessing-001.sql

@@ -19,10 +19,24 @@ BEGIN
     DECLARE @Proc_Step_Name VARCHAR(200)= '';
 	DECLARE @Dataflow_Name VARCHAR(200) = 'Event Metric Cleanup POST-Processing';
 	DECLARE @Package_Name VARCHAR(200) = 'sp_event_metric_cleanup_postprocessing';
+    DECLARE @lock_resource NVARCHAR(255) = N'RTR:scheduled:event-metric-cleanup';
+    DECLARE @lock_result INT;
+    DECLARE @lock_acquired BIT = 0;
 
     BEGIN TRY
-
 
+        EXEC @lock_result = sys.sp_getapplock
+            @Resource = @lock_resource,
+            @LockMode = N'Exclusive',
+            @LockOwner = N'Session',
+            @LockTimeout = 0;
+
+        IF @lock_result < 0
+        BEGIN
+            RETURN -2;
+        END;
+
+        SET @lock_acquired = 1;
         SET @Proc_Step_Name = 'SP_Start';
 
         INSERT INTO dbo.job_flow_log ( batch_id
@@ -206,14 +220,27 @@ BEGIN
         INSERT INTO [dbo].[job_flow_log] 
 		(batch_id, [Dataflow_Name], [package_Name], [Status_Type], [step_number], [step_name], [row_count])
         VALUES (@batch_id, @Dataflow_Name, @Package_Name, 'COMPLETE', 999, @Proc_Step_name, @RowCount_no);
-
+
+        EXEC sys.sp_releaseapplock
+            @Resource = @lock_resource,
+            @LockOwner = N'Session';
+
+        RETURN 1;
+
 -------------------------------------------------------------------------------------------
     END TRY
 
     BEGIN CATCH
 
         IF @@TRANCOUNT > 0   ROLLBACK TRANSACTION;
 
+        IF @lock_acquired = 1
+        BEGIN
+            EXEC sys.sp_releaseapplock
+                @Resource = @lock_resource,
+                @LockOwner = N'Session';
+        END;
+
         -- Construct the error message string with all details:
             DECLARE @FullErrorMessage VARCHAR(8000) =
                 'Error Number: ' + CAST(ERROR_NUMBER() AS VARCHAR(10)) + CHAR(13) + CHAR(10) +  -- Carriage return and line feed for new lines

reporting-pipeline-service/src/main/resources/db/changelog/migrations/v7.13/rdb/routines/371-sp_lab100_cleanup.sql

@@ -22,8 +22,23 @@ AS
       DECLARE @Proc_Step_Name VARCHAR(200)= '';
       DECLARE @Dataflow_Name VARCHAR(200) = 'Lab 100 Cleanup';
       DECLARE @Package_Name VARCHAR(200) = 'sp_lab100_cleanup';
+      DECLARE @lock_resource NVARCHAR(255) = N'RTR:scheduled:lab100-cleanup';
+      DECLARE @lock_result INT;
+      DECLARE @lock_acquired BIT = 0;
 
       BEGIN try
+          EXEC @lock_result = sys.sp_getapplock
+              @Resource = @lock_resource,
+              @LockMode = N'Exclusive',
+              @LockOwner = N'Session',
+              @LockTimeout = 0;
+
+          IF @lock_result < 0
+          BEGIN
+              RETURN -2;
+          END;
+
+          SET @lock_acquired = 1;
           SET @Proc_Step_Name = 'SP_Start';
 
           INSERT INTO dbo.job_flow_log
@@ -139,13 +154,26 @@ AS
 
 
         COMMIT TRANSACTION;
+
+        EXEC sys.sp_releaseapplock
+            @Resource = @lock_resource,
+            @LockOwner = N'Session';
+
+        RETURN 1;
       ---------------------------------------------------------------------------------------------------------
       END try
 
       BEGIN catch
           IF @@TRANCOUNT > 0
             ROLLBACK TRANSACTION;
 
+          IF @lock_acquired = 1
+          BEGIN
+              EXEC sys.sp_releaseapplock
+                  @Resource = @lock_resource,
+                  @LockOwner = N'Session';
+          END;
+
           -- Construct the error message string with all details:
           DECLARE @FullErrorMessage VARCHAR(8000) = 'Error Number: '
             + Cast(Error_number() AS VARCHAR(10))

reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/integration/unit/CleanupStatusRepositoryTest.java

@@ -0,0 +1,22 @@
+package gov.cdc.nbs.report.pipeline.integration.unit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+
+class CleanupStatusRepositoryTest extends UnitTest {
+
+  @Autowired private PostProcRepository postProcRepository;
+
+  @Test
+  void returnsExplicitSuccessForEventMetricCleanup() {
+    assertEquals(1, postProcRepository.executeEventMetricCleanup());
+  }
+
+  @Test
+  void returnsExplicitSuccessForLab100Cleanup() {
+    assertEquals(1, postProcRepository.executeLab100Cleanup());
+  }
+}

reporting-pipeline-service/src/test/java/gov/cdc/nbs/report/pipeline/integration/unit/ScheduledCleanupAppLockConcurrencyTest.java

@@ -0,0 +1,134 @@
+package gov.cdc.nbs.report.pipeline.integration.unit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTimeout;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import gov.cdc.nbs.report.pipeline.postprocessing.repository.PostProcRepository;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.function.Supplier;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+
+@Execution(ExecutionMode.SAME_THREAD)
+class ScheduledCleanupAppLockConcurrencyTest extends UnitTest {
+
+  private static final String EVENT_METRIC_LOCK = "RTR:scheduled:event-metric-cleanup";
+  private static final String EVENT_METRIC_PROCEDURE = "sp_event_metric_cleanup_postprocessing";
+  private static final String LAB100_LOCK = "RTR:scheduled:lab100-cleanup";
+  private static final String LAB100_PROCEDURE = "sp_lab100_cleanup";
+  private static final Duration SKIP_TIMEOUT = Duration.ofSeconds(2);
+
+  @Autowired private PostProcRepository postProcRepository;
+
+  @Value("${spring.datasource.admin.url}")
+  private String adminJdbcUrl;
+
+  @Value("${spring.datasource.admin.username}")
+  private String adminJdbcUser;
+
+  @Value("${spring.datasource.admin.password}")
+  private String adminJdbcPassword;
+
+  @Test
+  void eventMetricCleanupSkipsWithoutWritingNormalJobFlowLogsWhenLocked() throws SQLException {
+    assertSkipsWhenLocked(
+        EVENT_METRIC_LOCK, EVENT_METRIC_PROCEDURE, postProcRepository::executeEventMetricCleanup);
+  }
+
+  @Test
+  void lab100CleanupSkipsWithoutWritingNormalJobFlowLogsWhenLocked() throws SQLException {
+    assertSkipsWhenLocked(LAB100_LOCK, LAB100_PROCEDURE, postProcRepository::executeLab100Cleanup);
+  }
+
+  @Test
+  void cleanupLocksDoNotBlockTheOtherCleanupProcedure() throws SQLException {
+    try (Connection eventMetricLockOwner = adminConnection()) {
+      acquireLock(eventMetricLockOwner, EVENT_METRIC_LOCK);
+      try {
+        assertEquals(1, postProcRepository.executeLab100Cleanup());
+      } finally {
+        releaseLock(eventMetricLockOwner, EVENT_METRIC_LOCK);
+      }
+    }
+
+    try (Connection lab100LockOwner = adminConnection()) {
+      acquireLock(lab100LockOwner, LAB100_LOCK);
+      try {
+        assertEquals(1, postProcRepository.executeEventMetricCleanup());
+      } finally {
+        releaseLock(lab100LockOwner, LAB100_LOCK);
+      }
+    }
+  }
+
+  private void assertSkipsWhenLocked(
+      String lockResource, String procedureName, Supplier<Integer> scheduledProcedure)
+      throws SQLException {
+    try (Connection lockOwner = adminConnection()) {
+      acquireLock(lockOwner, lockResource);
+      try {
+        long jobFlowLogCount = normalJobFlowLogCount(lockOwner, procedureName);
+
+        assertTimeout(SKIP_TIMEOUT, () -> assertEquals(-2, scheduledProcedure.get()));
+
+        assertEquals(jobFlowLogCount, normalJobFlowLogCount(lockOwner, procedureName));
+      } finally {
+        releaseLock(lockOwner, lockResource);
+      }
+    }
+  }
+
+  private long normalJobFlowLogCount(Connection connection, String procedureName)
+      throws SQLException {
+    try (Statement statement = connection.createStatement();
+        ResultSet result =
+            statement.executeQuery(
+                "SELECT COUNT(*) FROM dbo.job_flow_log WHERE package_name = '"
+                    + procedureName
+                    + "' AND status_type IN ('START', 'COMPLETE')")) {
+      result.next();
+      return result.getLong(1);
+    }
+  }
+
+  private void acquireLock(Connection connection, String lockResource) throws SQLException {
+    assertTrue(lockResult(connection, lockResource) >= 0);
+  }
+
+  private int lockResult(Connection connection, String lockResource) throws SQLException {
+    try (Statement statement = connection.createStatement();
+        ResultSet result =
+            statement.executeQuery(
+                "DECLARE @lock_result int; EXEC @lock_result = sys.sp_getapplock "
+                    + "@Resource = N'"
+                    + lockResource
+                    + "', @LockMode = N'Exclusive', @LockOwner = N'Session', @LockTimeout = 0; "
+                    + "SELECT @lock_result")) {
+      result.next();
+      return result.getInt(1);
+    }
+  }
+
+  private void releaseLock(Connection connection, String lockResource) throws SQLException {
+    try (Statement statement = connection.createStatement()) {
+      statement.execute(
+          "EXEC sys.sp_releaseapplock @Resource = N'"
+              + lockResource
+              + "', @LockOwner = N'Session'");
+    }
+  }
+
+  private Connection adminConnection() throws SQLException {
+    String rdbJdbcUrl = adminJdbcUrl.replaceAll("databaseName=[^;]+", "databaseName=RDB_MODERN");
+    return DriverManager.getConnection(rdbJdbcUrl, adminJdbcUser, adminJdbcPassword);
+  }
+}