fix(concurrency): batch-2 hazards C1, H1-H7, M1-M9 (17 fixes)

obp-api/src/main/scala/code/accountaccessrequest/AccountAccessRequest.scala

@@ -1,9 +1,10 @@
 package code.accountaccessrequest
 
 import java.util.Date
+import code.api.util.ErrorMessages
 import code.util.{MappedUUID, UUIDString}
 import com.openbankproject.commons.model.enums.AccountAccessRequestStatus
-import net.liftweb.common.{Box, Full}
+import net.liftweb.common.{Box, Failure, Full}
 import net.liftweb.mapper._
 import net.liftweb.util.Helpers.tryo
 
@@ -90,13 +91,12 @@ object MappedAccountAccessRequestProvider extends AccountAccessRequestProvider {
     checkerComment: String
   ): Box[AccountAccessRequestTrait] = {
     AccountAccessRequest.find(By(AccountAccessRequest.AccountAccessRequestId, accountAccessRequestId)).flatMap { request =>
-      tryo {
-        request
-          .Status(status)
-          .CheckerUserId(checkerUserId)
-          .CheckerComment(checkerComment)
-          .saveMe()
-      }
+      // Atomic guarded transition: an access request is actioned once, from INITIATED. The loser of a
+      // concurrent approve/decline gets 0 rows -> Failure, instead of silently overwriting the decision.
+      val rows = code.bankconnectors.DoobieBusinessStatusQueries.conditionalAccountAccessRequestStatus(
+        request.id.get, AccountAccessRequestStatus.INITIATED.toString, status, checkerUserId, checkerComment)
+      if (rows == 1) AccountAccessRequest.find(By(AccountAccessRequest.AccountAccessRequestId, accountAccessRequestId))
+      else Failure(ErrorMessages.AccountAccessRequestStatusNotInitiated)
     }
   }
 }

obp-api/src/main/scala/code/accountapplication/MappedAccountApplication.scala

@@ -35,7 +35,16 @@ object MappedAccountApplicationProvider extends AccountApplicationProvider {
        match {
         case Full(accountApplication) if(accountApplication.status == "ACCEPTED") =>
           Failure(s"${ErrorMessages.AccountApplicationAlreadyAccepted} Current Account-Application-Id($accountApplicationId)")
-        case Full(accountApplication)  => tryo{accountApplication.mStatus(status).saveMe()}
+        case Full(accountApplication)  =>
+          // Optimistic CAS: transition only if the status hasn't changed since we loaded it. Two
+          // concurrent updates that both read the same status can no longer both write — the loser
+          // (0 rows) gets a Failure instead of silently overwriting the winner's decision.
+          val rows = code.bankconnectors.DoobieBusinessStatusQueries.conditionalAccountApplicationStatus(
+            accountApplication.id.get, accountApplication.status, status)
+          if (rows == 1) MappedAccountApplication.find(By(MappedAccountApplication.mAccountApplicationId, accountApplicationId))
+          // Use the generic update-failure code here: the concurrent winner may have written any
+          // status (ACCEPTED or REJECTED), so the "already accepted" message would be misleading.
+          else Failure(s"${ErrorMessages.UpdateAccountApplicationStatusError} Status changed concurrently. Current Account-Application-Id($accountApplicationId)")
         case Empty  => Failure(s"${ErrorMessages.AccountApplicationNotFound} Current Account-Application-Id($accountApplicationId)") 
         case _  => Failure(ErrorMessages.UnknownError) 
       }    

obp-api/src/main/scala/code/actorsystem/ObpActorSystem.scala

@@ -10,8 +10,10 @@ import com.typesafe.config.ConfigFactory
 object ObpActorSystem extends MdcLoggable {
 
   val props_hostname = Helper.getHostname
-  var obpActorSystem: ActorSystem = _
-  var northSideAkkaConnectorActorSystem: ActorSystem = _
+  // @volatile so the single assignment of each actor system is visible to all reader threads
+  // (the JVM memory model does not guarantee visibility of a non-volatile write across threads).
+  @volatile var obpActorSystem: ActorSystem = _
+  @volatile var northSideAkkaConnectorActorSystem: ActorSystem = _
 
   def startLocalActorSystem() = localActorSystem
 
@@ -22,12 +24,19 @@ object ObpActorSystem extends MdcLoggable {
     obpActorSystem = ActorSystem.create(s"ObpActorSystem_${props_hostname}", ConfigFactory.load(ConfigFactory.parseString(localConf)))
     obpActorSystem
   }
-
+
+  // synchronized double-checked init so concurrent callers start the connector system exactly once.
   def startNorthSideAkkaConnectorActorSystem(): ActorSystem = {
-    logger.info("Starting North Side Akka Connector actor system")
-    val localConf = AkkaConnectorActorConfig.localConf
-    logger.info(localConf)
-    northSideAkkaConnectorActorSystem = ActorSystem.create(s"SouthSideAkkaConnector_${props_hostname}", ConfigFactory.load(ConfigFactory.parseString(localConf)))
+    if (northSideAkkaConnectorActorSystem == null) {
+      synchronized {
+        if (northSideAkkaConnectorActorSystem == null) {
+          logger.info("Starting North Side Akka Connector actor system")
+          val localConf = AkkaConnectorActorConfig.localConf
+          logger.info(localConf)
+          northSideAkkaConnectorActorSystem = ActorSystem.create(s"SouthSideAkkaConnector_${props_hostname}", ConfigFactory.load(ConfigFactory.parseString(localConf)))
+        }
+      }
+    }
     northSideAkkaConnectorActorSystem
   }
 }

obp-api/src/main/scala/code/actorsystem/ObpLookupSystem.scala

@@ -16,14 +16,20 @@ object ObpLookupSystem extends ObpLookupSystem {
 }
 
 trait ObpLookupSystem extends MdcLoggable {
-  var obpLookupSystem: ActorSystem = null
+  // @volatile + synchronized double-checked init: without it two threads can both see null,
+  // both build an ActorSystem (resource leak), and a reader can observe a stale null.
+  @volatile var obpLookupSystem: ActorSystem = null
   val props_hostname = Helper.getHostname
 
   def init (): ActorSystem = {
-    if (obpLookupSystem == null ) {
-      val system = ActorSystem("ObpLookupSystem", ConfigFactory.load(ConfigFactory.parseString(ObpActorConfig.lookupConf)))
-      logger.info(ObpActorConfig.lookupConf)
-      obpLookupSystem = system
+    if (obpLookupSystem == null) {
+      synchronized {
+        if (obpLookupSystem == null) {
+          val system = ActorSystem("ObpLookupSystem", ConfigFactory.load(ConfigFactory.parseString(ObpActorConfig.lookupConf)))
+          logger.info(ObpActorConfig.lookupConf)
+          obpLookupSystem = system
+        }
+      }
     }
     obpLookupSystem
   }