Improved CreateProcess reliability by waiting for Confirmation. (#1930)

* Improved CreateProcess reliability by waiting for Confirmation.

* Compatability patch

---------

Co-authored-by: ku76uh <malte.heuser@ing.com>

Author: mltheuser

core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_index/ProcessIndex.scala

@@ -3,7 +3,7 @@ package com.ing.baker.runtime.akka.actor.process_index
 import akka.actor.{ActorRef, NoSerializationVerificationNeeded, Props}
 import akka.cluster.sharding.ShardRegion.Passivate
 import akka.event.{DiagnosticLoggingAdapter, Logging}
-import akka.pattern.{BackoffOpts, BackoffSupervisor, ask}
+import akka.pattern.{BackoffOpts, BackoffSupervisor, ask, pipe}
 import akka.persistence._
 import akka.sensors.actor.PersistentActorMetrics
 import cats.data.{EitherT, OptionT}
@@ -40,6 +40,7 @@ import com.typesafe.config.Config
 import scala.collection.mutable
 import scala.concurrent.duration._
 import scala.concurrent.{Await, ExecutionContext, Future}
+import scala.util.control.NonFatal
 import scala.util.{Failure, Success}
 
 
@@ -128,6 +129,10 @@ class ProcessIndex(recipeInstanceIdleTimeout: Option[FiniteDuration],
 
   override val log: DiagnosticLoggingAdapter = Logging.getLogger(logSource = this)
 
+  // --- Internal messages for the asynchronous initialization flow ---
+  private case class InitializationConfirmed(originalSender: ActorRef, recipeId: String, recipeInstanceId: String, compiledRecipe: CompiledRecipe, createdTime: Long) extends NoSerializationVerificationNeeded
+  private case class InitializationRejected(originalSender: ActorRef, recipeInstanceId: String, cause: Throwable) extends NoSerializationVerificationNeeded
+
   private val startTime = System.currentTimeMillis()
 
   override def preStart(): Unit = {
@@ -342,6 +347,60 @@ class ProcessIndex(recipeInstanceIdleTimeout: Option[FiniteDuration],
     } yield (transition, jobId)
   }
 
+  /**
+   * Creates the Initialize command for a new process instance.
+   *
+   * @return A tuple containing the Initialize command and the creation timestamp.
+   */
+  private def createInitializationCommand(
+                                           recipeId: String,
+                                           recipeInstanceId: String,
+                                           recipeInstanceMetadata: Map[String, String],
+                                           compiledRecipe: CompiledRecipe): (Initialize, Long) = {
+
+    val createdTime = System.currentTimeMillis()
+
+    val ingredientsMap =
+      if (recipeInstanceMetadata.isEmpty) Map.empty[String, Value]
+      else Map(RecipeInstanceMetadataName -> com.ing.baker.types.Converters.toValue(recipeInstanceMetadata))
+
+    val processState = RecipeInstanceState(
+      recipeId = recipeId,
+      recipeInstanceId = recipeInstanceId,
+      ingredients = ingredientsMap,
+      recipeInstanceMetadata = recipeInstanceMetadata,
+      events = List.empty)
+
+    val initializeCmd = Initialize(compiledRecipe.initialMarking, processState)
+
+    (initializeCmd, createdTime)
+  }
+
+  /**
+   * Handles the successful initialization of a process instance.
+   *
+   * This involves persisting the ActorCreated event, updating the in-memory index,
+   * logging the creation event, and replying to the original sender.
+   */
+  private def handleSuccessfulInitialization(
+                                              originalSender: ActorRef,
+                                              recipeId: String,
+                                              recipeInstanceId: String,
+                                              compiledRecipe: CompiledRecipe,
+                                              createdTime: Long): Unit = {
+
+    persistWithSnapshot(ActorCreated(recipeId, recipeInstanceId, createdTime)) { _ =>
+      // This callback runs after the event is successfully persisted.
+      val actorMetadata = ActorMetadata(recipeId, recipeInstanceId, createdTime, Active)
+      index += recipeInstanceId -> actorMetadata
+
+      val creationEvent = RecipeInstanceCreated(createdTime, recipeId, compiledRecipe.name, recipeInstanceId)
+      LogAndSendEvent.recipeInstanceCreated(creationEvent, context.system.eventStream)
+
+      originalSender ! Initialized(Marking.empty)
+    }
+  }
+
   override def receiveCommand: Receive =  {
     case SaveSnapshotSuccess(metadata) =>
       log.debug("Snapshot saved & cleaning old processes")
@@ -390,72 +449,48 @@ class ProcessIndex(recipeInstanceIdleTimeout: Option[FiniteDuration],
       }
 
     case CreateProcess(recipeId, recipeInstanceId, recipeInstanceMetadata) =>
-      context.child(recipeInstanceId) match {
-        case None if !index.contains(recipeInstanceId) =>
-
-          // First check if the recipe exists
-          getCompiledRecipe(recipeId) match {
-            case Some(compiledRecipe) =>
-
-              val createdTime = System.currentTimeMillis()
+      val originalSender = sender()
 
-              // this persists the fact that we created a process instance
-              persistWithSnapshot(ActorCreated(recipeId, recipeInstanceId, createdTime)) { _ =>
+      index.get(recipeInstanceId) match {
+        case Some(metadata) if metadata.isDeleted =>
+          originalSender ! ProcessDeleted(recipeInstanceId)
 
-                // after that we actually create the ProcessInstance actor
-                val processState = RecipeInstanceState(
-                  recipeId = recipeId,
-                  recipeInstanceId = recipeInstanceId,
-                  ingredients =
-                    if (recipeInstanceMetadata.isEmpty) Map.empty[String, Value]
-                    else Map(RecipeInstanceMetadataName -> com.ing.baker.types.Converters.toValue(recipeInstanceMetadata)),
-                  recipeInstanceMetadata = recipeInstanceMetadata,
-                  events = List.empty)
-                val initializeCmd = Initialize(compiledRecipe.initialMarking, processState)
+        case Some(_) =>
+          originalSender ! ProcessAlreadyExists(recipeInstanceId)
 
-                //TODO ensure the initialiseCMD is accepted before we add it ot the index
+        case None =>
+          getCompiledRecipe(recipeId) match {
+            case Some(compiledRecipe) =>
+              val processActor = context.child(recipeInstanceId).getOrElse {
                 createProcessActor(recipeInstanceId, compiledRecipe)
-                  .forward(initializeCmd)
-
-                val actorMetadata = ActorMetadata(recipeId, recipeInstanceId, createdTime, Active)
+              }
 
-                LogAndSendEvent.recipeInstanceCreated(
-                  RecipeInstanceCreated(System.currentTimeMillis(), recipeId, compiledRecipe.name, recipeInstanceId),
-                  context.system.eventStream)
+              val (initializeCmd, createdTime) = createInitializationCommand(
+                recipeId, recipeInstanceId, recipeInstanceMetadata, compiledRecipe)
 
-                index += recipeInstanceId -> actorMetadata
-              }
+              (processActor ? initializeCmd)(processInquireTimeout).map {
+                  case _: Initialized | _: AlreadyInitialized =>
+                    InitializationConfirmed(originalSender, recipeId, recipeInstanceId, compiledRecipe, createdTime)
+                  case other =>
+                    val err = new IllegalStateException(s"ProcessInstance for $recipeInstanceId replied with unexpected message: $other")
+                    InitializationRejected(originalSender, recipeInstanceId, err)
+                }
+                .recover { case NonFatal(e) => InitializationRejected(originalSender, recipeInstanceId, e) }
+                .pipeTo(self)
 
             case None =>
-              sender() ! NoRecipeFound(recipeId)
-          }
-        case _ if index.get(recipeInstanceId).exists(_.isDeleted) =>
-          sender() ! ProcessDeleted(recipeInstanceId)
-        case None =>
-          //Temporary solution for the situation that the initializeCmd is not send in the original Bake
-          getCompiledRecipe(recipeId) match {
-            case Some(compiledRecipe) =>
-              val processState = RecipeInstanceState(recipeId, recipeInstanceId, Map.empty[String, Value], recipeInstanceMetadata, List.empty)
-              val initializeCmd = Initialize(compiledRecipe.initialMarking, processState)
-              createProcessActor(recipeInstanceId, compiledRecipe) ! initializeCmd
-              sender() ! ProcessAlreadyExists(recipeInstanceId)
-            case None =>
-              //Kept the ProcessAlreadyExists since this was the original error
-              sender() ! ProcessAlreadyExists(recipeInstanceId)
-          }
-        case Some(actorRef: ActorRef) =>
-          //Temporary solution for the situation that the initializeCmd is not send in the original Bake
-          getCompiledRecipe(recipeId) match {
-            case Some(compiledRecipe) =>
-              val processState = RecipeInstanceState(recipeId, recipeInstanceId, Map.empty[String, Value], recipeInstanceMetadata, List.empty)
-              val initializeCmd = Initialize(compiledRecipe.initialMarking, processState)
-              actorRef ! initializeCmd
-              sender() ! ProcessAlreadyExists(recipeInstanceId)
-            case None =>
-              sender() ! NoRecipeFound(recipeId)
+              originalSender ! NoRecipeFound(recipeId)
           }
       }
 
+    case msg: InitializationConfirmed =>
+      handleSuccessfulInitialization(
+        msg.originalSender, msg.recipeId, msg.recipeInstanceId, msg.compiledRecipe, msg.createdTime)
+
+    case msg: InitializationRejected =>
+      log.error(msg.cause, s"Initialization of process ${msg.recipeInstanceId} failed.")
+      msg.originalSender ! akka.actor.Status.Failure(msg.cause)
+
     case command@ProcessEvent(recipeInstanceId, event, correlationId, _, _) =>
       run ({ responseHandler =>
         for {