Fix the bug that the ProcessInstance would passivate on retry interaction if that interaction took too long. (#1936)

* Fix for passivating the instance on interaction retry.

---------

Co-authored-by: Tim Linschoten <tim.linschoten@ing.com>

Author: Tim-Linschoten

core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_instance/ProcessInstance.scala

@@ -577,6 +577,7 @@ class ProcessInstance(
           // execute the job immediately if there is no timeout
           if (timeout == 0) {
             executeJob(job, originalSender)
+            updateState(updatedInstance, scheduledRetries)
           }
           else {
             // schedule the retry

core/akka-runtime/src/test/scala/com/ing/baker/runtime/akka/actor/process_instance/ProcessInstanceSpec.scala

@@ -4,6 +4,7 @@ import akka.actor.{ActorRef, ActorSystem, PoisonPill, Props, Terminated}
 import akka.event.DiagnosticLoggingAdapter
 import akka.testkit.{TestDuration, TestProbe}
 import akka.util.Timeout
+import cats.effect.IO
 import cats.effect.unsafe.IORuntime
 import com.ing.baker.il.failurestrategy.{BlockInteraction, FireEventAfterFailure, InteractionFailureStrategy, RetryWithIncrementalBackoff}
 import com.ing.baker.il.petrinet.{InteractionTransition, Place}
@@ -19,6 +20,7 @@ import com.ing.baker.runtime.akka.actor.process_instance.ProcessInstanceSpec._
 import com.ing.baker.runtime.akka.actor.process_instance.dsl.TestUtils.{PlaceMethods, place}
 import com.ing.baker.runtime.akka.actor.process_instance.dsl._
 import com.ing.baker.runtime.akka.actor.process_instance.internal.ExceptionStrategy.RetryWithDelay
+import com.ing.baker.runtime.akka.actor.process_instance.internal.Job
 import com.ing.baker.runtime.akka.actor.process_instance.{ProcessInstanceProtocol => protocol}
 import com.ing.baker.runtime.akka.internal.FatalInteractionException
 import com.ing.baker.runtime.akka.namedCachedThreadPool
@@ -30,6 +32,7 @@ import org.mockito.ArgumentMatchers.any
 import org.mockito.Mockito._
 import org.mockito.invocation.InvocationOnMock
 import org.mockito.stubbing.Answer
+import org.scalatest.BeforeAndAfterEach
 import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.time.{Milliseconds, Span}
@@ -56,9 +59,12 @@ trait TestSequenceNet extends SequenceNet {
 
 object ProcessInstanceSpec {
 
+  var countDownLatch = new java.util.concurrent.CountDownLatch(0)
+
   def Added(n: Int): EventInstance = EventInstance(n.toString, Map.empty)
 
   val testExecutionContext = namedCachedThreadPool(s"Baker.CachedThreadPool")
+  val interactionExecutorContext = namedCachedThreadPool(s"Baker.InteractionExecutor")
 
   val instanceSettings = Settings(
     executionContext = testExecutionContext,
@@ -79,7 +85,14 @@ object ProcessInstanceSpec {
       CompiledRecipe("name", UUID.randomUUID().toString, topology, Marking.empty, Seq.empty, Option.empty, Option.empty),
       settings,
       runtime,
-      delayedTransitionActor)
+      delayedTransitionActor) {
+        override def executeJobViaExecutor(job: Job[RecipeInstanceState], originalSender: ActorRef): Unit = {
+          IO.delay {
+            countDownLatch.await(10, TimeUnit.SECONDS)
+            super.executeJobViaExecutor(job, originalSender)
+          }.evalOn(interactionExecutorContext).unsafeRunAndForget()(IORuntime.global)
+        }
+      }
     )
   }
 
@@ -96,7 +109,7 @@ object ProcessInstanceSpec {
   }
 }
 
-class ProcessInstanceSpec extends AkkaTestBase("ProcessInstanceSpec") with ScalaFutures with MockitoSugar with Matchers {
+class ProcessInstanceSpec extends AkkaTestBase("ProcessInstanceSpec") with ScalaFutures with MockitoSugar with Matchers  with BeforeAndAfterEach {
 
   def dilatedMillis(millis: Long)(implicit system: ActorSystem): Long = FiniteDuration(millis, TimeUnit.MILLISECONDS).dilated.toMillis
 
@@ -107,6 +120,11 @@ class ProcessInstanceSpec extends AkkaTestBase("ProcessInstanceSpec") with Scala
     Thread.sleep(dilatedMillis(100))
   }
 
+  override def beforeEach(): Unit = {
+    countDownLatch = new java.util.concurrent.CountDownLatch(0)
+    super.beforeEach()
+  }
+
   "A persistent petri net actor" should {
 
     "Respond with an Initialized response after processing an Initialize command" in new TestSequenceNet {
@@ -190,7 +208,6 @@ class ProcessInstanceSpec extends AkkaTestBase("ProcessInstanceSpec") with Scala
     }
 
     "Be able to retry a failed (blocked) transition when requested" in new TestSequenceNet {
-
       val counter = new AtomicInteger(0)
 
       override val sequence = Seq(
@@ -207,25 +224,99 @@ class ProcessInstanceSpec extends AkkaTestBase("ProcessInstanceSpec") with Scala
       expectMsgClass(classOf[Initialized])
 
       actor ! FireTransition(transitionId = 1, input = null)
-
       expectMsgClass(classOf[TransitionFailed])
 
+      // Validate the instance is in the correct state
       actor ! GetState
-
       val state: InstanceState = expectMsgClass(classOf[InstanceState])
-
       state.jobs.size shouldBe 1
-
       val (jobId, jobState) = state.jobs.head
-
       jobState.exceptionState should matchPattern {
         case Some(ExceptionState(_, _, BlockTransition)) =>
       }
 
+      // Set the Countdown latch so that the job is only finished when we want.
+      countDownLatch = new java.util.concurrent.CountDownLatch(1)
+
+      // Send the OverrideExceptionStrategy message
       actor ! OverrideExceptionStrategy(jobId, protocol.ExceptionStrategy.RetryWithDelay(0))
 
-      // expect that the failure is resolved
+      // Validate job is executing
+      actor ! GetState
+      val newState: InstanceState = expectMsgClass(classOf[InstanceState])
+      newState.jobs.size shouldBe 1
+      val (newJobId, newJobState) = newState.jobs.head
+      newJobId shouldBe jobId
+      newJobState.exceptionState should matchPattern {
+        case Some(ExceptionState(_, _, com.ing.baker.runtime.akka.actor.process_instance.ProcessInstanceProtocol.ExceptionStrategy.RetryWithDelay(0))) =>
+      }
+
+      // Allow the job to finish
+      countDownLatch.countDown()
+
+      // Validate the transition is fired successfully after retry
+      expectMsgPF() { case TransitionFired(_, 1, _, _, _, _, _) => }
+
+      // Validate no job is left open
+      actor ! GetState
+      val finalState: InstanceState = expectMsgClass(classOf[InstanceState])
+      finalState.jobs.size shouldBe 0
+    }
+
+    "Be able to retry a failed (blocked) transition when requested with delay" in new TestSequenceNet {
+      val counter = new AtomicInteger(0)
+
+      override val sequence = Seq(
+        transition() { _ =>
+          if (counter.getAndIncrement() == 0)
+            throw new RuntimeException("t1 failed!")
+          else
+            Added(1)
+        })
+
+      val actor = createProcessInstance(petriNet, runtime)
+
+      actor ! Initialize(initialMarking, RecipeInstanceState(UUID.randomUUID().toString, UUID.randomUUID().toString, Map.empty, Map.empty, Seq.empty))
+      expectMsgClass(classOf[Initialized])
+
+      actor ! FireTransition(transitionId = 1, input = null)
+      expectMsgClass(classOf[TransitionFailed])
+
+      // Validate the instance is in the correct state
+      actor ! GetState
+      val state: InstanceState = expectMsgClass(classOf[InstanceState])
+      state.jobs.size shouldBe 1
+      val (jobId, jobState) = state.jobs.head
+      jobState.exceptionState should matchPattern {
+        case Some(ExceptionState(_, _, BlockTransition)) =>
+      }
+
+      // Set the Countdown latch so that the job is only finished when we want.
+      countDownLatch = new java.util.concurrent.CountDownLatch(1)
+
+      // Send the OverrideExceptionStrategy message
+      actor ! OverrideExceptionStrategy(jobId, protocol.ExceptionStrategy.RetryWithDelay(1000))
+
+      // Validate job is executing
+      actor ! GetState
+      val newState: InstanceState = expectMsgClass(classOf[InstanceState])
+      newState.jobs.size shouldBe 1
+      val (newJobId, newJobState) = newState.jobs.head
+      newJobId shouldBe jobId
+      newJobState.exceptionState should matchPattern {
+        case Some(ExceptionState(_, _, com.ing.baker.runtime.akka.actor.process_instance.ProcessInstanceProtocol.ExceptionStrategy.RetryWithDelay(1000))) =>
+      }
+
+      // Allow the job to finish
+      countDownLatch.countDown()
+
+      // Validate the transition is fired successfully after retry
       expectMsgPF() { case TransitionFired(_, 1, _, _, _, _, _) => }
+
+      // Validate no job is left open
+      actor ! GetState
+      val finalState: InstanceState = expectMsgClass(classOf[InstanceState])
+      finalState.jobs.size shouldBe 0
     }
 
     "Be able to resolve a failed (blocked) transition when requested" in new TestSequenceNet {