Handle all finished periodic scenarios waiting for reschedule instead…

… of one (#7391)
TouK · Jan 7, 2025 · 2623924 · 2623924
1 parent 223f484
commit 2623924
Show file tree

Hide file tree

Showing 2 changed files with 58 additions and 1 deletion.
diff --git a/...scala/pl/touk/nussknacker/engine/management/periodic/db/PeriodicProcessesRepository.scala b/...scala/pl/touk/nussknacker/engine/management/periodic/db/PeriodicProcessesRepository.scala
@@ -325,7 +325,11 @@ class SlickPeriodicProcessesRepository(
     getLatestDeploymentsForEachSchedule(
       processesHavingDeploymentsWithMatchingStatus,
       deploymentsPerScheduleMaxCount = 1
-    ).map(_.values.headOption.getOrElse(SchedulesState(Map.empty)))
+    ).map(schedulesForProcessNames =>
+      SchedulesState(
+        schedulesForProcessNames.values.map(_.schedules).foldLeft(Map.empty[ScheduleId, ScheduleData])(_ ++ _)
+      )
+    )
   }
 
   override def getLatestDeploymentsForActiveSchedules(

diff --git a/...l/touk/nussknacker/engine/management/periodic/PeriodicProcessServiceIntegrationTest.scala b/...l/touk/nussknacker/engine/management/periodic/PeriodicProcessServiceIntegrationTest.scala
@@ -235,6 +235,7 @@ class PeriodicProcessServiceIntegrationTest
     )
     service.handleFinished.futureValue
 
+    // here we check that scenarios that not fired are still on the "toDeploy" list and finished are not on the list
     val toDeployAfterFinish = service.findToBeDeployed.futureValue
     toDeployAfterFinish.map(_.periodicProcess.processVersion.processName) should contain only every30MinutesProcessName
     service.deactivate(processName).futureValue
@@ -270,6 +271,58 @@ class PeriodicProcessServiceIntegrationTest
     )
   }
 
+  it should "handleFinished for all finished periodic scenarios waiting for reschedule" in withFixture() { f =>
+    val timeToTriggerCheck = startTime.plus(2, ChronoUnit.HOURS)
+    var currentTime        = startTime
+    def service            = f.periodicProcessService(currentTime)
+
+    service
+      .schedule(
+        cronEveryHour,
+        ProcessVersion.empty.copy(processName = ProcessName("first")),
+        sampleProcess,
+        randomProcessActionId
+      )
+      .futureValue
+    service
+      .schedule(
+        cronEveryHour,
+        ProcessVersion.empty.copy(processName = ProcessName("second")),
+        sampleProcess,
+        randomProcessActionId
+      )
+      .futureValue
+
+    currentTime = timeToTriggerCheck
+
+    // deploy all
+    service.findToBeDeployed.futureValue
+      .foreach(pp => service.deploy(pp).futureValue)
+
+    val stateAfterDeploy = service.getLatestDeploymentsForActiveSchedules(1).futureValue
+    stateAfterDeploy should have size 2
+
+    // finish all
+    stateAfterDeploy.values.foreach(schedulesState => {
+      val deployment = schedulesState.firstScheduleData.latestDeployments.head
+      f.delegateDeploymentManagerStub.setStateStatus(
+        processName,
+        SimpleStateStatus.Finished,
+        Some(deployment.id)
+      )
+    })
+    service.handleFinished.futureValue
+
+    // check all are rescheduled for next run
+    val stateAfterFinish = service.getLatestDeploymentsForActiveSchedules(1).futureValue
+    stateAfterFinish.values.foreach(schedulesState => {
+      val deployment = schedulesState.firstScheduleData.latestDeployments.head
+      deployment.state should matchPattern {
+        case PeriodicProcessDeploymentState(_, _, PeriodicProcessDeploymentStatus.Scheduled) =>
+      }
+    })
+  }
+
   it should "redeploy scenarios that failed on deploy" in withFixture(deploymentRetryConfig =
     DeploymentRetryConfig(deployMaxRetries = 1)
   ) { f =>